kallithea Changeset - 620c13a373c5

Changeset - 620c13a373c5

Parent rev.

Child rev.

[Not reviewed]

default

0 14 0

Mads Kiilerich - 6 years ago 2019-12-26 15:17:51
mads@kiilerich.com

Grafted from: 3d7166c2b89e

py3: trivial renaming of unicode to str

14 files changed with 35 insertions and 35 deletions:

kallithea/controllers/api/__init__.py

kallithea/lib/celerylib/__init__.py

kallithea/lib/helpers.py

kallithea/lib/recaptcha.py

kallithea/lib/utils2.py

kallithea/lib/vcs/backends/base.py

kallithea/lib/vcs/backends/hg/repository.py

kallithea/lib/vcs/subprocessio.py

kallithea/model/db.py

kallithea/model/gist.py

kallithea/model/notification.py

kallithea/tests/vcs/test_git.py

kallithea/tests/vcs/test_hg.py

kallithea/tests/vcs/test_inmemchangesets.py

0 comments (0 inline, 0 general)

kallithea/controllers/api/__init__.py

➞

Show inline comments

@@ @@ -37,233 +37,233 @@ from webob.exc import HTTPError, HTTPExc @@
 from kallithea.lib import ext_json
 from kallithea.lib.auth import AuthUser
 from kallithea.lib.base import _get_ip_addr as _get_ip
 from kallithea.lib.base import get_path_info
 from kallithea.lib.utils2 import ascii_bytes
 from kallithea.model.db import User
 log = logging.getLogger('JSONRPC')
 class JSONRPCError(BaseException):
     def __init__(self, message):
         self.message = message
         super(JSONRPCError, self).__init__()
     def __str__(self):
         return self.message
 class JSONRPCErrorResponse(Response, HTTPException):
     """
     Generate a Response object with a JSON-RPC error body
     """
     def __init__(self, message=None, retid=None, code=None):
         HTTPException.__init__(self, message, self)
         Response.__init__(self,
                           json_body=dict(id=retid, result=None, error=message),
                           status=code,
                           content_type='application/json')
 class JSONRPCController(TGController):
     """
      A WSGI-speaking JSON-RPC controller class
      See the specification:
      <http://json-rpc.org/wiki/specification>`.
      Valid controller return values should be json-serializable objects.
      Sub-classes should catch their exceptions and raise JSONRPCError
      if they want to pass meaningful errors to the client.
      """
     def _get_ip_addr(self, environ):
         return _get_ip(environ)
     def _get_method_args(self):
         """
         Return `self._rpc_args` to dispatched controller method
         chosen by __call__
         """
         return self._rpc_args
     def _dispatch(self, state, remainder=None):
         """
         Parse the request body as JSON, look up the method on the
         controller and if it exists, dispatch to it.
         """
         # Since we are here we should respond as JSON
         response.content_type = 'application/json'
         environ = state.request.environ
         start = time.time()
         ip_addr = self._get_ip_addr(environ)
         self._req_id = None
         if 'CONTENT_LENGTH' not in environ:
             log.debug("No Content-Length")
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message="No Content-Length in request")
         else:
             length = environ['CONTENT_LENGTH'] or 0
             length = int(environ['CONTENT_LENGTH'])
             log.debug('Content-Length: %s', length)
         if length == 0:
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message="Content-Length is 0")
         raw_body = environ['wsgi.input'].read(length)
         try:
             json_body = ext_json.loads(raw_body)
         except ValueError as e:
             # catch JSON errors Here
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message="JSON parse error ERR:%s RAW:%r"
                                                 % (e, raw_body))
         # check AUTH based on API key
         try:
             self._req_api_key = json_body['api_key']
             self._req_id = json_body['id']
             self._req_method = json_body['method']
             self._request_params = json_body['args']
             if not isinstance(self._request_params, dict):
                 self._request_params = {}
             log.debug('method: %s, params: %s',
                       self._req_method, self._request_params)
         except KeyError as e:
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message='Incorrect JSON query missing %s' % e)
         # check if we can find this session using api_key
         try:
             u = User.get_by_api_key(self._req_api_key)
             auth_user = AuthUser.make(dbuser=u, ip_addr=ip_addr)
             if auth_user is None:
                 raise JSONRPCErrorResponse(retid=self._req_id,
                                            message='Invalid API key')
         except Exception as e:
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message='Invalid API key')
         request.authuser = auth_user
         request.ip_addr = ip_addr
         self._error = None
         try:
             self._func = self._find_method()
         except AttributeError as e:
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message=str(e))
         # now that we have a method, add self._req_params to
         # self.kargs and dispatch control to WGIController
         argspec = inspect.getfullargspec(self._func)
         arglist = argspec.args[1:]
         argtypes = [type(arg) for arg in argspec.defaults or []]
         default_empty = type(NotImplemented)
         # kw arguments required by this method
         func_kwargs = dict(itertools.zip_longest(reversed(arglist), reversed(argtypes),
                                                   fillvalue=default_empty))
         # This attribute will need to be first param of a method that uses
         # api_key, which is translated to instance of user at that name
         USER_SESSION_ATTR = 'apiuser'
         # get our arglist and check if we provided them as args
         for arg, default in func_kwargs.items():
             if arg == USER_SESSION_ATTR:
                 # USER_SESSION_ATTR is something translated from API key and
                 # this is checked before so we don't need validate it
                 continue
             # skip the required param check if it's default value is
             # NotImplementedType (default_empty)
             if default == default_empty and arg not in self._request_params:
                 raise JSONRPCErrorResponse(
                     retid=self._req_id,
                     message='Missing non optional `%s` arg in JSON DATA' % arg,
+                )
         extra = set(self._request_params).difference(func_kwargs)
         if extra:
             raise JSONRPCErrorResponse(
                 retid=self._req_id,
                 message='Unknown %s arg in JSON DATA' %
                         ', '.join('`%s`' % arg for arg in extra),
+            )
         self._rpc_args = {}
         self._rpc_args.update(self._request_params)
         self._rpc_args['action'] = self._req_method
         self._rpc_args['environ'] = environ
         log.info('IP: %s Request to %s time: %.3fs' % (
             self._get_ip_addr(environ),
             get_path_info(environ), time.time() - start)
+        )
         state.set_action(self._rpc_call, [])
         state.set_params(self._rpc_args)
         return state
     def _rpc_call(self, action, environ, **rpc_args):
         """
         Call the specified RPC Method
         """
         raw_response = ''
         try:
             raw_response = getattr(self, action)(**rpc_args)
             if isinstance(raw_response, HTTPError):
                 self._error = str(raw_response)
         except JSONRPCError as e:
-            self._error = unicode(e)
+            self._error = str(e)
         except Exception as e:
             log.error('Encountered unhandled exception: %s',
                       traceback.format_exc(),)
             json_exc = JSONRPCError('Internal server error')
-            self._error = unicode(json_exc)
+            self._error = str(json_exc)
         if self._error is not None:
             raw_response = None
         response = dict(id=self._req_id, result=raw_response, error=self._error)
         try:
             return ascii_bytes(ext_json.dumps(response))
         except TypeError as e:
             log.error('API FAILED. Error encoding response for %s %s: %s\n%s', action, rpc_args, e, traceback.format_exc())
             return ascii_bytes(ext_json.dumps(
                 dict(
                     id=self._req_id,
                     result=None,
                     error="Error encoding response",
+                )
             ))
     def _find_method(self):
         """
         Return method named by `self._req_method` in controller if able
         """
         log.debug('Trying to find JSON-RPC method: %s', self._req_method)
         if self._req_method.startswith('_'):
             raise AttributeError("Method not allowed")
         try:
             func = getattr(self, self._req_method, None)
         except UnicodeEncodeError:
             raise AttributeError("Problem decoding unicode in requested "
                                  "method name.")
         if isinstance(func, types.MethodType):
             return func
         else:
             raise AttributeError("No such method: %s" % (self._req_method,))

kallithea/lib/celerylib/__init__.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.celerylib
 ~~~~~~~~~~~~~~~~~~~~~~~
 celery libs 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 27, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import os
 from hashlib import md5
 from decorator import decorator
 from tg import config
 from kallithea import CELERY_EAGER, CELERY_ON
 from kallithea.lib.pidlock import DaemonLock, LockHeld
 from kallithea.lib.utils2 import safe_bytes
 from kallithea.model import meta
 log = logging.getLogger(__name__)
 class FakeTask(object):
     """Fake a sync result to make it look like a finished task"""
     def __init__(self, result):
         self.result = result
     def failed(self):
         return False
     traceback = None # if failed
     task_id = None
 def task(f_org):
     """Wrapper of celery.task.task, running async if CELERY_ON
     """
     if CELERY_ON:
         def f_async(*args, **kwargs):
             log.info('executing %s task', f_org.__name__)
             try:
                 f_org(*args, **kwargs)
             finally:
                 log.info('executed %s task', f_org.__name__)
         f_async.__name__ = f_org.__name__
         from kallithea.lib import celerypylons
         runner = celerypylons.task(ignore_result=True)(f_async)
         def f_wrapped(*args, **kwargs):
             t = runner.apply_async(args=args, kwargs=kwargs)
             log.info('executing task %s in async mode - id %s', f_org, t.task_id)
             return t
     else:
         def f_wrapped(*args, **kwargs):
             log.info('executing task %s in sync', f_org.__name__)
             try:
                 result = f_org(*args, **kwargs)
             except Exception as e:
                 log.error('exception executing sync task %s in sync: %r', f_org.__name__, e)
                 raise # TODO: return this in FakeTask as with async tasks?
             return FakeTask(result)
     return f_wrapped
 def __get_lockkey(func, *fargs, **fkwargs):
     params = list(fargs)
     params.extend(['%s-%s' % ar for ar in fkwargs.items()])
     func_name = str(func.__name__) if hasattr(func, '__name__') else str(func)
     lockkey = 'task_%s.lock' % \
-        md5(safe_bytes(func_name + '-' + '-'.join(unicode(x) for x in params))).hexdigest()
+        md5(safe_bytes(func_name + '-' + '-'.join(str(x) for x in params))).hexdigest()
     return lockkey
 def locked_task(func):
     def __wrapper(func, *fargs, **fkwargs):
         lockkey = __get_lockkey(func, *fargs, **fkwargs)
         lockkey_path = config.get('cache_dir') or config['app_conf']['cache_dir']  # Backward compatibility for TurboGears < 2.4
         log.info('running task with lockkey %s', lockkey)
         try:
             l = DaemonLock(os.path.join(lockkey_path, lockkey))
             ret = func(*fargs, **fkwargs)
             l.release()
             return ret
         except LockHeld:
             log.info('LockHeld')
             return 'Task with key %s already running' % lockkey
     return decorator(__wrapper, func)
 def get_session():
     sa = meta.Session()
     return sa
 def dbsession(func):
     def __wrapper(func, *fargs, **fkwargs):
         try:
             ret = func(*fargs, **fkwargs)
             return ret
         finally:
             if CELERY_ON and not CELERY_EAGER:
                 meta.Session.remove()
     return decorator(__wrapper, func)

kallithea/lib/helpers.py

➞

Show inline comments

@@ @@ -271,389 +271,389 @@ class CodeHtmlFormatter(HtmlFormatter): @@
                     else:
                         if aln:
                             lines.append('<a href="#%s%d">%*d</a>' % (la, i, mw, i))
                         else:
                             lines.append('%*d' % (mw, i))
                 else:
                     lines.append('')
             ls = '\n'.join(lines)
         else:
             lines = []
             for i in range(fl, fl + lncount):
                 if i % st == 0:
                     if aln:
                         lines.append('<a href="#%s%d">%*d</a>' % (la, i, mw, i))
                     else:
                         lines.append('%*d' % (mw, i))
                 else:
                     lines.append('')
             ls = '\n'.join(lines)
         # in case you wonder about the seemingly redundant <div> here: since the
         # content in the other cell also is wrapped in a div, some browsers in
         # some configurations seem to mess up the formatting...
         if nocls:
             yield 0, ('<table class="%stable">' % self.cssclass +
                       '<tr><td><div class="linenodiv">'
                       '<pre>' + ls + '</pre></div></td>'
                       '<td id="hlcode" class="code">')
         else:
             yield 0, ('<table class="%stable">' % self.cssclass +
                       '<tr><td class="linenos"><div class="linenodiv">'
                       '<pre>' + ls + '</pre></div></td>'
                       '<td id="hlcode" class="code">')
         yield 0, ''.join(inner_lines)
         yield 0, '</td></tr></table>'
 _whitespace_re = re.compile(r'(\t)|( )(?=\n|</div>)')
 def _markup_whitespace(m):
     groups = m.groups()
     if groups[0]:
         return '<u>\t</u>'
     if groups[1]:
         return ' <i></i>'
 def markup_whitespace(s):
     return _whitespace_re.sub(_markup_whitespace, s)
 def pygmentize(filenode, **kwargs):
     """
     pygmentize function using pygments
     :param filenode:
     """
     lexer = get_custom_lexer(filenode.extension) or filenode.lexer
     return literal(markup_whitespace(
         code_highlight(safe_str(filenode.content), lexer, CodeHtmlFormatter(**kwargs))))
 def hsv_to_rgb(h, s, v):
     if s == 0.0:
         return v, v, v
     i = int(h * 6.0)  # XXX assume int() truncates!
     f = (h * 6.0) - i
     p = v * (1.0 - s)
     q = v * (1.0 - s * f)
     t = v * (1.0 - s * (1.0 - f))
     i = i % 6
     if i == 0:
         return v, t, p
     if i == 1:
         return q, v, p
     if i == 2:
         return p, v, t
     if i == 3:
         return p, q, v
     if i == 4:
         return t, p, v
     if i == 5:
         return v, p, q
 def gen_color(n=10000):
     """generator for getting n of evenly distributed colors using
     hsv color and golden ratio. It always return same order of colors
     :returns: RGB tuple
     """
     golden_ratio = 0.618033988749895
     h = 0.22717784590367374
     for _unused in range(n):
         h += golden_ratio
         h %= 1
         HSV_tuple = [h, 0.95, 0.95]
         RGB_tuple = hsv_to_rgb(*HSV_tuple)
         yield [str(int(x * 256)) for x in RGB_tuple]
 def pygmentize_annotation(repo_name, filenode, **kwargs):
     """
     pygmentize function for annotation
     :param filenode:
     """
     cgenerator = gen_color()
     color_dict = {}
     def get_color_string(cs):
         if cs in color_dict:
             col = color_dict[cs]
         else:
             col = color_dict[cs] = next(cgenerator)
         return "color: rgb(%s)! important;" % (', '.join(col))
     def url_func(changeset):
         author = escape(changeset.author)
         date = changeset.date
         message = escape(changeset.message)
         tooltip_html = ("<b>Author:</b> %s<br/>"
                         "<b>Date:</b> %s</b><br/>"
                         "<b>Message:</b> %s") % (author, date, message)
         lnk_format = show_id(changeset)
         uri = link_to(
                 lnk_format,
                 url('changeset_home', repo_name=repo_name,
                     revision=changeset.raw_id),
                 style=get_color_string(changeset.raw_id),
                 **{'data-toggle': 'popover',
                    'data-content': tooltip_html}
+              )
         uri += '\n'
         return uri
     return literal(markup_whitespace(annotate_highlight(filenode, url_func, **kwargs)))
 class _Message(object):
     """A message returned by ``pop_flash_messages()``.
     Converting the message to a string returns the message text. Instances
     also have the following attributes:
     * ``category``: the category specified when the message was created.
     * ``message``: the html-safe message text.
     """
     def __init__(self, category, message):
         self.category = category
         self.message = message
 def _session_flash_messages(append=None, clear=False):
     """Manage a message queue in tg.session: return the current message queue
     after appending the given message, and possibly clearing the queue."""
     key = 'flash'
     from tg import session
     if key in session:
         flash_messages = session[key]
     else:
         if append is None:  # common fast path - also used for clearing empty queue
             return []  # don't bother saving
         flash_messages = []
         session[key] = flash_messages
     if append is not None and append not in flash_messages:
         flash_messages.append(append)
     if clear:
         session.pop(key, None)
     session.save()
     return flash_messages
 def flash(message, category, logf=None):
     """
     Show a message to the user _and_ log it through the specified function
     category: notice (default), warning, error, success
     logf: a custom log function - such as log.debug
     logf defaults to log.info, unless category equals 'success', in which
     case logf defaults to log.debug.
     """
     assert category in ('error', 'success', 'warning'), category
     if hasattr(message, '__html__'):
         # render to HTML for storing in cookie
-        safe_message = unicode(message)
+        safe_message = str(message)
     else:
         # Apply str - the message might be an exception with __str__
         # Escape, so we can trust the result without further escaping, without any risk of injection
-        safe_message = html_escape(unicode(message))
+        safe_message = html_escape(str(message))
     if logf is None:
         logf = log.info
         if category == 'success':
             logf = log.debug
     logf('Flash %s: %s', category, safe_message)
     _session_flash_messages(append=(category, safe_message))
 def pop_flash_messages():
     """Return all accumulated messages and delete them from the session.
     The return value is a list of ``Message`` objects.
     """
     return [_Message(category, message) for category, message in _session_flash_messages(clear=True)]
 age = lambda x, y=False: _age(x, y)
 capitalize = lambda x: x.capitalize()
 email = author_email
 short_id = lambda x: x[:12]
 hide_credentials = lambda x: ''.join(credentials_filter(x))
 def show_id(cs):
     """
     Configurable function that shows ID
     by default it's r123:fffeeefffeee
     :param cs: changeset instance
     """
     from kallithea import CONFIG
     def_len = safe_int(CONFIG.get('show_sha_length', 12))
     show_rev = str2bool(CONFIG.get('show_revision_number', False))
     raw_id = cs.raw_id[:def_len]
     if show_rev:
         return 'r%s:%s' % (cs.revision, raw_id)
     else:
         return raw_id
 def fmt_date(date):
     if date:
         return date.strftime("%Y-%m-%d %H:%M:%S")
     return ""
 def is_git(repository):
     if hasattr(repository, 'alias'):
         _type = repository.alias
     elif hasattr(repository, 'repo_type'):
         _type = repository.repo_type
     else:
         _type = repository
     return _type == 'git'
 def is_hg(repository):
     if hasattr(repository, 'alias'):
         _type = repository.alias
     elif hasattr(repository, 'repo_type'):
         _type = repository.repo_type
     else:
         _type = repository
     return _type == 'hg'
 @cache_region('long_term', 'user_attr_or_none')
 def user_attr_or_none(author, show_attr):
     """Try to match email part of VCS committer string with a local user and return show_attr
     - or return None if user not found"""
     email = author_email(author)
     if email:
         from kallithea.model.db import User
         user = User.get_by_email(email, cache=True) # cache will only use sql_cache_short
         if user is not None:
             return getattr(user, show_attr)
     return None
 def email_or_none(author):
     """Try to match email part of VCS committer string with a local user.
     Return primary email of user, email part of the specified author name, or None."""
     if not author:
         return None
     email = user_attr_or_none(author, 'email')
     if email is not None:
         return email # always use user's main email address - not necessarily the one used to find user
     # extract email from the commit string
     email = author_email(author)
     if email:
         return email
     # No valid email, not a valid user in the system, none!
     return None
 def person(author, show_attr="username"):
     """Find the user identified by 'author', return one of the users attributes,
     default to the username attribute, None if there is no user"""
     from kallithea.model.db import User
     # if author is already an instance use it for extraction
     if isinstance(author, User):
         return getattr(author, show_attr)
     value = user_attr_or_none(author, show_attr)
     if value is not None:
         return value
     # Still nothing?  Just pass back the author name if any, else the email
     return author_name(author) or email(author)
 def person_by_id(id_, show_attr="username"):
     from kallithea.model.db import User
     # attr to return from fetched user
     person_getter = lambda usr: getattr(usr, show_attr)
     # maybe it's an ID ?
     if str(id_).isdigit() or isinstance(id_, int):
         id_ = int(id_)
         user = User.get(id_)
         if user is not None:
             return person_getter(user)
     return id_
 def boolicon(value):
     """Returns boolean value of a value, represented as small html image of true/false
     icons
     :param value: value
     """
     if value:
         return HTML.tag('i', class_="icon-ok")
     else:
         return HTML.tag('i', class_="icon-minus-circled")
 def action_parser(user_log, feed=False, parse_cs=False):
     """
     This helper will action_map the specified string action into translated
     fancy names with icons and links
     :param user_log: user log instance
     :param feed: use output for feeds (no html and fancy icons)
     :param parse_cs: parse Changesets into VCS instances
     """
     action = user_log.action
     action_params = ' '
     x = action.split(':')
     if len(x) > 1:
         action, action_params = x
     def get_cs_links():
         revs_limit = 3  # display this amount always
         revs_top_limit = 50  # show upto this amount of changesets hidden
         revs_ids = action_params.split(',')
         deleted = user_log.repository is None
         if deleted:
             return ','.join(revs_ids)
         repo_name = user_log.repository.repo_name
         def lnk(rev, repo_name):
             lazy_cs = False
             title_ = None
             url_ = '#'
             if isinstance(rev, BaseChangeset) or isinstance(rev, AttributeDict):
                 if rev.op and rev.ref_name:
                     if rev.op == 'delete_branch':
                         lbl = _('Deleted branch: %s') % rev.ref_name
                     elif rev.op == 'tag':
                         lbl = _('Created tag: %s') % rev.ref_name
                     else:
                         lbl = 'Unknown operation %s' % rev.op
                 else:
                     lazy_cs = True
                     lbl = rev.short_id[:8]
                     url_ = url('changeset_home', repo_name=repo_name,
                                revision=rev.raw_id)
             else:
                 # changeset cannot be found - it might have been stripped or removed
                 lbl = rev[:12]
                 title_ = _('Changeset %s not found') % lbl

kallithea/lib/recaptcha.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 import json
 import urllib.parse
 import urllib.request
 class RecaptchaResponse(object):
     def __init__(self, is_valid, error_code=None):
         self.is_valid = is_valid
         self.error_code = error_code
     def __repr__(self):
         return '<RecaptchaResponse:%s>' % (self.is_valid)
 def submit(g_recaptcha_response, private_key, remoteip):
     """
     Submits a reCAPTCHA request for verification. Returns RecaptchaResponse for the request
     g_recaptcha_response -- The value of g_recaptcha_response from the form
     private_key -- your reCAPTCHA private key
     remoteip -- the user's IP address
     """
     if not (g_recaptcha_response and len(g_recaptcha_response)):
         return RecaptchaResponse(is_valid=False, error_code='incorrect-captcha-sol')
     def encode_if_necessary(s):
-        if isinstance(s, unicode):
+        if isinstance(s, str):
             return s.encode('utf-8')
         return s
     params = urllib.parse.urlencode({
         'secret': encode_if_necessary(private_key),
         'remoteip': encode_if_necessary(remoteip),
         'response': encode_if_necessary(g_recaptcha_response),
     }).encode('ascii')
     req = urllib.request.Request(
         url="https://www.google.com/recaptcha/api/siteverify",
         data=params,
         headers={
             "Content-type": "application/x-www-form-urlencoded",
             "User-agent": "reCAPTCHA Python"
+        }
+    )
     httpresp = urllib.request.urlopen(req)
     return_values = json.loads(httpresp.read())
     httpresp.close()
     if not (isinstance(return_values, dict)):
         return RecaptchaResponse(is_valid=False, error_code='incorrect-captcha-sol')
     elif (("success" in return_values) and ((return_values["success"] is True) or (return_values["success"] == "true"))):
         return RecaptchaResponse(is_valid=True)
     elif (("error-codes" in return_values) and isinstance(return_values["error-codes"], list) and (len(return_values["error-codes"]) > 0)):
         return RecaptchaResponse(is_valid=False, error_code=return_values["error-codes"][0])
     else:
         return RecaptchaResponse(is_valid=False, error_code='incorrect-captcha-sol')

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.
 """
 from __future__ import print_function
 import binascii
 import datetime
 import json
 import os
 import pwd
 import re
 import time
 import urllib.parse
 import urlobject
 from tg.i18n import ugettext as _
 from tg.i18n import ungettext
 from webhelpers2.text import collapse, remove_formatting, strip_tags
 from kallithea.lib.vcs.utils import ascii_bytes, ascii_str, safe_bytes, safe_str  # re-export
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 def str2bool(_str):
     """
     returns True/False value from given string, it tries to translate the
     string into boolean
     :param _str: string value to translate into boolean
     :rtype: boolean
     :returns: boolean from given string
     """
     if _str is None:
         return False
     if _str in (True, False):
         return _str
     _str = str(_str).strip().lower()
     return _str in ('t', 'true', 'y', 'yes', 'on', '1')
 def aslist(obj, sep=None, strip=True):
     """
     Returns given string separated by sep as list
     :param obj:
     :param sep:
     :param strip:
     """
     if isinstance(obj, (str)):
         lst = obj.split(sep)
         if strip:
             lst = [v.strip() for v in lst]
         return lst
     elif isinstance(obj, (list, tuple)):
         return obj
     elif obj is None:
         return []
     else:
         return [obj]
 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 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: unicode
     :returns: unicode words describing age
     :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
     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
+    :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 Exception: # 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 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, str):
             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_usernames(text):
     r"""
     Returns list of (possible) usernames @mentioned in given text.
     >>> extract_mentioned_usernames('@1-2.a_X,@1234 not@not @ddd@not @n @ee @ff @gg, @gg;@hh @n\n@zz,')
     ['1-2.a_X', '1234', 'ddd', 'ee', 'ff', 'gg', 'gg', 'hh', 'zz']
     """
     return MENTIONS_REGEX.findall(text)
 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 not user.is_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 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)
 class HookEnvironmentError(Exception): pass
 def get_hook_environment():
     """
     Get hook context by deserializing the global KALLITHEA_EXTRAS environment
     variable.
     Called early in Git out-of-process hooks to get .ini config path so the
     basic environment can be configured properly. Also used in all hooks to get
     information about the action that triggered it.
     """
     try:
         kallithea_extras = os.environ['KALLITHEA_EXTRAS']
     except KeyError:
         raise HookEnvironmentError("Environment variable KALLITHEA_EXTRAS not found")
     extras = json.loads(kallithea_extras)
     for k in ['username', 'repository', 'scm', 'action', 'ip', 'config']:
         try:
             extras[k]
         except KeyError:
             raise HookEnvironmentError('Missing key %s in KALLITHEA_EXTRAS %s' % (k, extras))
     return AttributeDict(extras)
 def set_hook_environment(username, ip_addr, repo_name, repo_alias, action=None):
     """Prepare global context for running hooks by serializing data in the
     global KALLITHEA_EXTRAS environment variable.
     Most importantly, this allow Git hooks to do proper logging and updating of
     caches after pushes.
     Must always be called before anything with hooks are invoked.
     """
     from kallithea import CONFIG
     extras = {
         'ip': ip_addr, # used in log_push/pull_action action_logger
         'username': username,

kallithea/lib/vcs/backends/base.py

➞

Show inline comments

@@ @@ -153,388 +153,388 @@ class BaseRepository(object): @@
         """
         for revision in self.revisions:
             yield self.get_changeset(revision)
     def get_changesets(self, start=None, end=None, start_date=None,
                        end_date=None, branch_name=None, reverse=False, max_revisions=None):
         """
         Returns iterator of ``BaseChangeset`` objects from start to end,
         both inclusive.
         :param start: None or str
         :param end: None or str
         :param start_date:
         :param end_date:
         :param branch_name:
         :param reversed:
         """
         raise NotImplementedError
     def __getitem__(self, key):
         if isinstance(key, slice):
             return (self.get_changeset(rev) for rev in self.revisions[key])
         return self.get_changeset(key)
     def count(self):
         return len(self.revisions)
     def tag(self, name, user, revision=None, message=None, date=None, **opts):
         """
         Creates and returns a tag for the given ``revision``.
         :param name: name for new tag
         :param user: full username, i.e.: "Joe Doe <joe.doe@example.com>"
         :param revision: changeset id for which new tag would be created
         :param message: message of the tag's commit
         :param date: date of tag's commit
         :raises TagAlreadyExistError: if tag with same name already exists
         """
         raise NotImplementedError
     def remove_tag(self, name, user, message=None, date=None):
         """
         Removes tag with the given ``name``.
         :param name: name of the tag to be removed
         :param user: full username, i.e.: "Joe Doe <joe.doe@example.com>"
         :param message: message of the tag's removal commit
         :param date: date of tag's removal commit
         :raises TagDoesNotExistError: if tag with given name does not exists
         """
         raise NotImplementedError
     def get_diff(self, rev1, rev2, path=None, ignore_whitespace=False,
             context=3):
         """
         Returns (git like) *diff*, as plain text. Shows changes introduced by
         ``rev2`` since ``rev1``.
         :param rev1: Entry point from which diff is shown. Can be
           ``self.EMPTY_CHANGESET`` - in this case, patch showing all
           the changes since empty state of the repository until ``rev2``
         :param rev2: Until which revision changes should be shown.
         :param ignore_whitespace: If set to ``True``, would not show whitespace
           changes. Defaults to ``False``.
         :param context: How many lines before/after changed lines should be
           shown. Defaults to ``3``.
         """
         raise NotImplementedError
     # ========== #
     # COMMIT API #
     # ========== #
     @LazyProperty
     def in_memory_changeset(self):
         """
         Returns ``InMemoryChangeset`` object for this repository.
         """
         raise NotImplementedError
     def add(self, filenode, **kwargs):
         """
         Commit api function that will add given ``FileNode`` into this
         repository.
         :raises ``NodeAlreadyExistsError``: if there is a file with same path
           already in repository
         :raises ``NodeAlreadyAddedError``: if given node is already marked as
           *added*
         """
         raise NotImplementedError
     def remove(self, filenode, **kwargs):
         """
         Commit api function that will remove given ``FileNode`` into this
         repository.
         :raises ``EmptyRepositoryError``: if there are no changesets yet
         :raises ``NodeDoesNotExistError``: if there is no file with given path
         """
         raise NotImplementedError
     def commit(self, message, **kwargs):
         """
         Persists current changes made on this repository and returns newly
         created changeset.
         :raises ``NothingChangedError``: if no changes has been made
         """
         raise NotImplementedError
     def get_state(self):
         """
         Returns dictionary with ``added``, ``changed`` and ``removed`` lists
         containing ``FileNode`` objects.
         """
         raise NotImplementedError
     def get_config_value(self, section, name, config_file=None):
         """
         Returns configuration value for a given [``section``] and ``name``.
         :param section: Section we want to retrieve value from
         :param name: Name of configuration we want to retrieve
         :param config_file: A path to file which should be used to retrieve
           configuration from (might also be a list of file paths)
         """
         raise NotImplementedError
     def get_user_name(self, config_file=None):
         """
         Returns user's name from global configuration file.
         :param config_file: A path to file which should be used to retrieve
           configuration from (might also be a list of file paths)
         """
         raise NotImplementedError
     def get_user_email(self, config_file=None):
         """
         Returns user's email from global configuration file.
         :param config_file: A path to file which should be used to retrieve
           configuration from (might also be a list of file paths)
         """
         raise NotImplementedError
     # =========== #
     # WORKDIR API #
     # =========== #
     @LazyProperty
     def workdir(self):
         """
         Returns ``Workdir`` instance for this repository.
         """
         raise NotImplementedError
 class BaseChangeset(object):
     """
     Each backend should implement it's changeset representation.
     **Attributes**
         ``repository``
             repository object within which changeset exists
         ``raw_id``
             raw changeset representation (i.e. full 40 length sha for git
             backend)
         ``short_id``
             shortened (if apply) version of ``raw_id``; it would be simple
             shortcut for ``raw_id[:12]`` for git/mercurial backends or same
             as ``raw_id`` for subversion
         ``revision``
             revision number as integer
         ``files``
             list of ``FileNode`` (``Node`` with NodeKind.FILE) objects
         ``dirs``
             list of ``DirNode`` (``Node`` with NodeKind.DIR) objects
         ``nodes``
             combined list of ``Node`` objects
         ``author``
-            author of the changeset, as unicode
+            author of the changeset, as str
         ``message``
-            message of the changeset, as unicode
+            message of the changeset, as str
         ``parents``
             list of parent changesets
         ``last``
             ``True`` if this is last changeset in repository, ``False``
             otherwise; trying to access this attribute while there is no
             changesets would raise ``EmptyRepositoryError``
     """
     def __str__(self):
         return '<%s at %s:%s>' % (self.__class__.__name__, self.revision,
             self.short_id)
     def __repr__(self):
         return self.__str__()
     def __eq__(self, other):
         if type(self) is not type(other):
             return False
         return self.raw_id == other.raw_id
     def __json__(self, with_file_list=False):
         if with_file_list:
             return dict(
                 short_id=self.short_id,
                 raw_id=self.raw_id,
                 revision=self.revision,
                 message=self.message,
                 date=self.date,
                 author=self.author,
                 added=[el.path for el in self.added],
                 changed=[el.path for el in self.changed],
                 removed=[el.path for el in self.removed],
+            )
         else:
             return dict(
                 short_id=self.short_id,
                 raw_id=self.raw_id,
                 revision=self.revision,
                 message=self.message,
                 date=self.date,
                 author=self.author,
+            )
     @LazyProperty
     def last(self):
         if self.repository is None:
             raise ChangesetError("Cannot check if it's most recent revision")
         return self.raw_id == self.repository.revisions[-1]
     @LazyProperty
     def parents(self):
         """
         Returns list of parents changesets.
         """
         raise NotImplementedError
     @LazyProperty
     def children(self):
         """
         Returns list of children changesets.
         """
         raise NotImplementedError
     @LazyProperty
     def raw_id(self):
         """
         Returns raw string identifying this changeset.
         """
         raise NotImplementedError
     @LazyProperty
     def short_id(self):
         """
         Returns shortened version of ``raw_id`` attribute, as string,
         identifying this changeset, useful for web representation.
         """
         raise NotImplementedError
     @LazyProperty
     def revision(self):
         """
         Returns integer identifying this changeset.
         """
         raise NotImplementedError
     @LazyProperty
     def committer(self):
         """
         Returns Committer for given commit
         """
         raise NotImplementedError
     @LazyProperty
     def committer_name(self):
         """
         Returns Author name for given commit
         """
         return author_name(self.committer)
     @LazyProperty
     def committer_email(self):
         """
         Returns Author email address for given commit
         """
         return author_email(self.committer)
     @LazyProperty
     def author(self):
         """
         Returns Author for given commit
         """
         raise NotImplementedError
     @LazyProperty
     def author_name(self):
         """
         Returns Author name for given commit
         """
         return author_name(self.author)
     @LazyProperty
     def author_email(self):
         """
         Returns Author email address for given commit
         """
         return author_email(self.author)
     def get_file_mode(self, path):
         """
         Returns stat mode of the file at the given ``path``.
         """
         raise NotImplementedError
     def get_file_content(self, path):
         """
         Returns content of the file at the given ``path``.
         """
         raise NotImplementedError
     def get_file_size(self, path):
         """
         Returns size of the file at the given ``path``.
         """
         raise NotImplementedError
     def get_file_changeset(self, path):
         """
         Returns last commit of the file at the given ``path``.
         """
         raise NotImplementedError
     def get_file_history(self, path):
         """
         Returns history of file as reversed list of ``Changeset`` objects for
         which file at given ``path`` has been modified.
         """
         raise NotImplementedError
     def get_nodes(self, path):
         """
         Returns combined ``DirNode`` and ``FileNode`` objects list representing
         state of changeset at the given ``path``.
         :raises ``ChangesetError``: if node at the given ``path`` is not
           instance of ``DirNode``
         """
         raise NotImplementedError
     def get_node(self, path):
         """
         Returns ``Node`` object from the given ``path``.
         :raises ``NodeDoesNotExistError``: if there is no node at the given
           ``path``
         """
         raise NotImplementedError
     def fill_archive(self, stream=None, kind='tgz', prefix=None):
         """
         Fills up given stream.
         :param stream: file like object.
         :param kind: one of following: ``zip``, ``tar``, ``tgz``
             or ``tbz2``. Default: ``tgz``.

kallithea/lib/vcs/backends/hg/repository.py

➞

Show inline comments

@@ @@ -239,379 +239,379 @@ class MercurialRepository(BaseRepository @@
         return [ascii_str(self._repo[x].hex()) for x in self._repo.filtered(b'visible').changelog.revs()]
     def get_diff(self, rev1, rev2, path='', ignore_whitespace=False,
                   context=3):
         """
         Returns (git like) *diff*, as plain text. Shows changes introduced by
         ``rev2`` since ``rev1``.
         :param rev1: Entry point from which diff is shown. Can be
           ``self.EMPTY_CHANGESET`` - in this case, patch showing all
           the changes since empty state of the repository until ``rev2``
         :param rev2: Until which revision changes should be shown.
         :param ignore_whitespace: If set to ``True``, would not show whitespace
           changes. Defaults to ``False``.
         :param context: How many lines before/after changed lines should be
           shown. Defaults to ``3``. If negative value is passed-in, it will be
           set to ``0`` instead.
         """
         # Negative context values make no sense, and will result in
         # errors. Ensure this does not happen.
         if context < 0:
             context = 0
         if hasattr(rev1, 'raw_id'):
             rev1 = getattr(rev1, 'raw_id')
         if hasattr(rev2, 'raw_id'):
             rev2 = getattr(rev2, 'raw_id')
         # Check if given revisions are present at repository (may raise
         # ChangesetDoesNotExistError)
         if rev1 != self.EMPTY_CHANGESET:
             self.get_changeset(rev1)
         self.get_changeset(rev2)
         if path:
             file_filter = mercurial.match.exact(path)
         else:
             file_filter = None
         return b''.join(mercurial.patch.diff(self._repo, rev1, rev2, match=file_filter,
                           opts=mercurial.mdiff.diffopts(git=True,
                                         showfunc=True,
                                         ignorews=ignore_whitespace,
                                         context=context)))
     @classmethod
     def _check_url(cls, url, repoui=None):
         """
         Function will check given url and try to verify if it's a valid
         link. Sometimes it may happened that mercurial will issue basic
         auth request that can cause whole API to hang when used from python
         or other external calls.
         On failures it'll raise urllib2.HTTPError, exception is also thrown
         when the return code is non 200
         """
         # check first if it's not an local url
         if os.path.isdir(url) or url.startswith(b'file:'):
             return True
         if url.startswith(b'ssh:'):
             # in case of invalid uri or authentication issues, sshpeer will
             # throw an exception.
             mercurial.sshpeer.instance(repoui or mercurial.ui.ui(), url, False).lookup(b'tip')
             return True
         url_prefix = None
         if b'+' in url[:url.find(b'://')]:
             url_prefix, url = url.split(b'+', 1)
         handlers = []
         url_obj = mercurial.util.url(url)
         test_uri, authinfo = url_obj.authinfo()
         url_obj.passwd = b'*****'
         cleaned_uri = str(url_obj)
         if authinfo:
             # create a password manager
             passmgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
             passmgr.add_password(*authinfo)
             handlers.extend((mercurial.url.httpbasicauthhandler(passmgr),
                              mercurial.url.httpdigestauthhandler(passmgr)))
         o = urllib.request.build_opener(*handlers)
         o.addheaders = [('Content-Type', 'application/mercurial-0.1'),
                         ('Accept', 'application/mercurial-0.1')]
         req = urllib.request.Request(
             "%s?%s" % (
                 test_uri,
                 urllib.parse.urlencode({
                     'cmd': 'between',
                     'pairs': "%s-%s" % ('0' * 40, '0' * 40),
                 })
             ))
         try:
             resp = o.open(req)
             if resp.code != 200:
                 raise Exception('Return Code is not 200')
         except Exception as e:
             # means it cannot be cloned
             raise urllib.error.URLError("[%s] org_exc: %s" % (cleaned_uri, e))
         if not url_prefix: # skip svn+http://... (and git+... too)
             # now check if it's a proper hg repo
             try:
                 mercurial.httppeer.instance(repoui or mercurial.ui.ui(), url, False).lookup(b'tip')
             except Exception as e:
                 raise urllib.error.URLError(
                     "url [%s] does not look like an hg repo org_exc: %s"
                     % (cleaned_uri, e))
         return True
     def _get_repo(self, create, src_url=None, update_after_clone=False):
         """
         Function will check for mercurial repository in given path and return
         a localrepo object. If there is no repository in that path it will
         raise an exception unless ``create`` parameter is set to True - in
         that case repository would be created and returned.
         If ``src_url`` is given, would try to clone repository from the
         location at given clone_point. Additionally it'll make update to
         working copy accordingly to ``update_after_clone`` flag
         """
         try:
             if src_url:
                 url = safe_bytes(self._get_url(src_url))
                 opts = {}
                 if not update_after_clone:
                     opts.update({'noupdate': True})
                 MercurialRepository._check_url(url, self.baseui)
                 mercurial.commands.clone(self.baseui, url, safe_bytes(self.path), **opts)
                 # Don't try to create if we've already cloned repo
                 create = False
             return mercurial.localrepo.instance(self.baseui, safe_bytes(self.path), create=create)
         except (mercurial.error.Abort, mercurial.error.RepoError) as err:
             if create:
                 msg = "Cannot create repository at %s. Original error was %s" \
                     % (self.name, err)
             else:
                 msg = "Not valid repository at %s. Original error was %s" \
                     % (self.name, err)
             raise RepositoryError(msg)
     @LazyProperty
     def in_memory_changeset(self):
         return MercurialInMemoryChangeset(self)
     @LazyProperty
     def description(self):
         _desc = self._repo.ui.config(b'web', b'description', None, untrusted=True)
         return safe_str(_desc or b'unknown')
     @LazyProperty
     def contact(self):
         return safe_str(mercurial.hgweb.common.get_contact(self._repo.ui.config)
                             or b'Unknown')
     @LazyProperty
     def last_change(self):
         """
         Returns last change made on this repository as datetime object
         """
         return date_fromtimestamp(self._get_mtime(), makedate()[1])
     def _get_mtime(self):
         try:
             return time.mktime(self.get_changeset().date.timetuple())
         except RepositoryError:
             # fallback to filesystem
             cl_path = os.path.join(self.path, '.hg', "00changelog.i")
             st_path = os.path.join(self.path, '.hg', "store")
             if os.path.exists(cl_path):
                 return os.stat(cl_path).st_mtime
             else:
                 return os.stat(st_path).st_mtime
     def _get_revision(self, revision):
         """
         Given any revision identifier, returns a 40 char string with revision hash.
         :param revision: str or int or None
         """
         if self._empty:
             raise EmptyRepositoryError("There are no changesets yet")
         if revision in [-1, None]:
             revision = b'tip'
-        elif isinstance(revision, unicode):
+        elif isinstance(revision, str):
             revision = safe_bytes(revision)
         try:
             if isinstance(revision, int):
                 return ascii_str(self._repo[revision].hex())
             return ascii_str(mercurial.scmutil.revsymbol(self._repo, revision).hex())
         except (IndexError, ValueError, mercurial.error.RepoLookupError, TypeError):
             msg = "Revision %r does not exist for %s" % (safe_str(revision), self.name)
             raise ChangesetDoesNotExistError(msg)
         except (LookupError, ):
             msg = "Ambiguous identifier `%s` for %s" % (safe_str(revision), self.name)
             raise ChangesetDoesNotExistError(msg)
     def get_ref_revision(self, ref_type, ref_name):
         """
         Returns revision number for the given reference.
         """
         if ref_type == 'rev' and not ref_name.strip('0'):
             return self.EMPTY_CHANGESET
         # lookup up the exact node id
         _revset_predicates = {
                 'branch': 'branch',
                 'book': 'bookmark',
                 'tag': 'tag',
                 'rev': 'id',
+            }
         # avoid expensive branch(x) iteration over whole repo
         rev_spec = "%%s & %s(%%s)" % _revset_predicates[ref_type]
         try:
             revs = self._repo.revs(rev_spec, ref_name, ref_name)
         except LookupError:
             msg = "Ambiguous identifier %s:%s for %s" % (ref_type, ref_name, self.name)
             raise ChangesetDoesNotExistError(msg)
         except mercurial.error.RepoLookupError:
             msg = "Revision %s:%s does not exist for %s" % (ref_type, ref_name, self.name)
             raise ChangesetDoesNotExistError(msg)
         if revs:
             revision = revs.last()
         else:
             # TODO: just report 'not found'?
             revision = ref_name
         return self._get_revision(revision)
     def _get_archives(self, archive_name='tip'):
         allowed = self.baseui.configlist(b"web", b"allow_archive",
                                          untrusted=True)
         for name, ext in [(b'zip', '.zip'), (b'gz', '.tar.gz'), (b'bz2', '.tar.bz2')]:
             if name in allowed or self._repo.ui.configbool(b"web",
                                                            b"allow" + name,
                                                            untrusted=True):
                 yield {"type": safe_str(name), "extension": ext, "node": archive_name}
     def _get_url(self, url):
         """
         Returns normalized url. If schema is not given, fall back to
         filesystem (``file:///``) schema.
         """
         if url != 'default' and '://' not in url:
             url = "file:" + urllib.request.pathname2url(url)
         return url
     def get_changeset(self, revision=None):
         """
         Returns ``MercurialChangeset`` object representing repository's
         changeset at the given ``revision``.
         """
         return MercurialChangeset(repository=self, revision=self._get_revision(revision))
     def get_changesets(self, start=None, end=None, start_date=None,
                        end_date=None, branch_name=None, reverse=False, max_revisions=None):
         """
         Returns iterator of ``MercurialChangeset`` objects from start to end
         (both are inclusive)
         :param start: None, str, int or mercurial lookup format
         :param end:  None, str, int or mercurial lookup format
         :param start_date:
         :param end_date:
         :param branch_name:
         :param reversed: return changesets in reversed order
         """
         start_raw_id = self._get_revision(start)
         start_pos = None if start is None else self.revisions.index(start_raw_id)
         end_raw_id = self._get_revision(end)
         end_pos = None if end is None else self.revisions.index(end_raw_id)
         if start_pos is not None and end_pos is not None and start_pos > end_pos:
             raise RepositoryError("Start revision '%s' cannot be "
                                   "after end revision '%s'" % (start, end))
         if branch_name and branch_name not in self.allbranches:
             msg = "Branch %r not found in %s" % (branch_name, self.name)
             raise BranchDoesNotExistError(msg)
         if end_pos is not None:
             end_pos += 1
         # filter branches
         filter_ = []
         if branch_name:
             filter_.append(b'branch("%s")' % safe_bytes(branch_name))
         if start_date:
             filter_.append(b'date(">%s")' % safe_bytes(str(start_date)))
         if end_date:
             filter_.append(b'date("<%s")' % safe_bytes(str(end_date)))
         if filter_ or max_revisions:
             if filter_:
                 revspec = b' and '.join(filter_)
             else:
                 revspec = b'all()'
             if max_revisions:
                 revspec = b'limit(%s, %d)' % (revspec, max_revisions)
             revisions = mercurial.scmutil.revrange(self._repo, [revspec])
         else:
             revisions = self.revisions
         # this is very much a hack to turn this into a list; a better solution
         # would be to get rid of this function entirely and use revsets
         revs = list(revisions)[start_pos:end_pos]
         if reverse:
             revs.reverse()
         return CollectionGenerator(self, revs)
     def pull(self, url):
         """
         Tries to pull changes from external location.
         """
         other = mercurial.hg.peer(self._repo, {}, safe_bytes(self._get_url(url)))
         try:
             mercurial.exchange.pull(self._repo, other, heads=None, force=None)
         except mercurial.error.Abort as err:
             # Propagate error but with vcs's type
             raise RepositoryError(str(err))
     @LazyProperty
     def workdir(self):
         """
         Returns ``Workdir`` instance for this repository.
         """
         return MercurialWorkdir(self)
     def get_config_value(self, section, name=None, config_file=None):
         """
         Returns configuration value for a given [``section``] and ``name``.
         :param section: Section we want to retrieve value from
         :param name: Name of configuration we want to retrieve
         :param config_file: A path to file which should be used to retrieve
           configuration from (might also be a list of file paths)
         """
         if config_file is None:
             config_file = []
         elif isinstance(config_file, str):
             config_file = [config_file]
         config = self._repo.ui
         if config_file:
             config = mercurial.ui.ui()
             for path in config_file:
                 config.readconfig(safe_bytes(path))
         value = config.config(safe_bytes(section), safe_bytes(name))
         return value if value is None else safe_str(value)
     def get_user_name(self, config_file=None):
         """
         Returns user's name from global configuration file.
         :param config_file: A path to file which should be used to retrieve
           configuration from (might also be a list of file paths)
         """
         username = self.get_config_value('ui', 'username', config_file=config_file)
         if username:
             return author_name(username)
         return None
     def get_user_email(self, config_file=None):
         """
         Returns user's email from global configuration file.
         :param config_file: A path to file which should be used to retrieve
           configuration from (might also be a list of file paths)
         """
         username = self.get_config_value('ui', 'username', config_file=config_file)
         if username:
             return author_email(username)
         return None

kallithea/lib/vcs/subprocessio.py

➞

Show inline comments

@@ @@ -97,328 +97,328 @@ class InputStreamChunker(threading.Threa @@
         self.go.set()
     def stop(self):
         self.go.clear()
         self.EOF.set()
         try:
             # this is not proper, but is done to force the reader thread let
             # go of the input because, if successful, .close() will send EOF
             # down the pipe.
             self.source.close()
         except:
             pass
     def run(self):
         s = self.source
         t = self.target
         cs = self.chunk_size
         ccm = self.chunk_count_max
         kr = self.keep_reading
         da = self.data_added
         go = self.go
         try:
             b = s.read(cs)
         except ValueError:
             b = ''
         while b and go.is_set():
             if len(t) > ccm:
                 kr.clear()
                 kr.wait(2)
                 # # this only works on 2.7.x and up
                 # if not kr.wait(10):
                 #     raise Exception("Timed out while waiting for input to be read.")
                 # instead we'll use this
                 if len(t) > ccm + 3:
                     raise IOError(
                         "Timed out while waiting for input from subprocess.")
             t.append(b)
             da.set()
             try:
                 b = s.read(cs)
             except ValueError: # probably "I/O operation on closed file"
                 b = ''
         self.EOF.set()
         da.set()  # for cases when done but there was no input.
 class BufferedGenerator(object):
     """
     Class behaves as a non-blocking, buffered pipe reader.
     Reads chunks of data (through a thread)
     from a blocking pipe, and attaches these to an array (Deque) of chunks.
     Reading is halted in the thread when max chunks is internally buffered.
     The .next() may operate in blocking or non-blocking fashion by yielding
     '' if no data is ready
     to be sent or by not returning until there is some data to send
     When we get EOF from underlying source pipe we raise the marker to raise
     StopIteration after the last chunk of data is yielded.
     """
     def __init__(self, source, buffer_size=65536, chunk_size=4096,
                  starting_values=None, bottomless=False):
         starting_values = starting_values or []
         if bottomless:
             maxlen = int(buffer_size / chunk_size)
         else:
             maxlen = None
         self.data = collections.deque(starting_values, maxlen)
         self.worker = InputStreamChunker(source, self.data, buffer_size,
                                          chunk_size)
         if starting_values:
             self.worker.data_added.set()
         self.worker.start()
     ####################
     # Generator's methods
     ####################
     def __iter__(self):
         return self
     def __next__(self):
         while not len(self.data) and not self.worker.EOF.is_set():
             self.worker.data_added.clear()
             self.worker.data_added.wait(0.2)
         if len(self.data):
             self.worker.keep_reading.set()
             return bytes(self.data.popleft())
         elif self.worker.EOF.is_set():
             raise StopIteration
     def throw(self, type, value=None, traceback=None):
         if not self.worker.EOF.is_set():
             raise type(value)
     def start(self):
         self.worker.start()
     def stop(self):
         self.worker.stop()
     def close(self):
         try:
             self.worker.stop()
             self.throw(GeneratorExit)
         except (GeneratorExit, StopIteration):
             pass
     ####################
     # Threaded reader's infrastructure.
     ####################
     @property
     def input(self):
         return self.worker.w
     @property
     def data_added_event(self):
         return self.worker.data_added
     @property
     def data_added(self):
         return self.worker.data_added.is_set()
     @property
     def reading_paused(self):
         return not self.worker.keep_reading.is_set()
     @property
     def done_reading_event(self):
         """
         Done_reading does not mean that the iterator's buffer is empty.
         Iterator might have done reading from underlying source, but the read
         chunks might still be available for serving through .next() method.
         :returns: An threading.Event class instance.
         """
         return self.worker.EOF
     @property
     def done_reading(self):
         """
         Done_reading does not mean that the iterator's buffer is empty.
         Iterator might have done reading from underlying source, but the read
         chunks might still be available for serving through .next() method.
         :returns: An Bool value.
         """
         return self.worker.EOF.is_set()
     @property
     def length(self):
         """
         returns int.
         This is the length of the queue of chunks, not the length of
         the combined contents in those chunks.
         __len__() cannot be meaningfully implemented because this
         reader is just flying through a bottomless pit content and
         can only know the length of what it already saw.
         If __len__() on WSGI server per PEP 3333 returns a value,
         the response's length will be set to that. In order not to
         confuse WSGI PEP3333 servers, we will not implement __len__
         at all.
         """
         return len(self.data)
     def prepend(self, x):
         self.data.appendleft(x)
     def append(self, x):
         self.data.append(x)
     def extend(self, o):
         self.data.extend(o)
     def __getitem__(self, i):
         return self.data[i]
 class SubprocessIOChunker(object):
     """
     Processor class wrapping handling of subprocess IO.
     In a way, this is a "communicate()" replacement with a twist.
     - We are multithreaded. Writing in and reading out, err are all sep threads.
     - We support concurrent (in and out) stream processing.
-    - The output is not a stream. It's a queue of read string (bytes, not unicode)
+    - The output is not a stream. It's a queue of read string (bytes, not str)
       chunks. The object behaves as an iterable. You can "for chunk in obj:" us.
     - We are non-blocking in more respects than communicate()
       (reading from subprocess out pauses when internal buffer is full, but
        does not block the parent calling code. On the flip side, reading from
        slow-yielding subprocess may block the iteration until data shows up. This
        does not block the parallel inpipe reading occurring parallel thread.)
     The purpose of the object is to allow us to wrap subprocess interactions into
     an iterable that can be passed to a WSGI server as the application's return
     value. Because of stream-processing-ability, WSGI does not have to read ALL
     of the subprocess's output and buffer it, before handing it to WSGI server for
     HTTP response. Instead, the class initializer reads just a bit of the stream
     to figure out if error occurred or likely to occur and if not, just hands the
     further iteration over subprocess output to the server for completion of HTTP
     response.
     The real or perceived subprocess error is trapped and raised as one of
     EnvironmentError family of exceptions
     Example usage:
     #    try:
     #        answer = SubprocessIOChunker(
     #            cmd,
     #            input,
     #            buffer_size = 65536,
     #            chunk_size = 4096
     #            )
     #    except (EnvironmentError) as e:
     #        print str(e)
     #        raise e
+    #
     #    return answer
     """
     def __init__(self, cmd, inputstream=None, buffer_size=65536,
                  chunk_size=4096, starting_values=None, **kwargs):
         """
         Initializes SubprocessIOChunker
         :param cmd: A Subprocess.Popen style "cmd". Can be string or array of strings
         :param inputstream: (Default: None) A file-like, string, or file pointer.
         :param buffer_size: (Default: 65536) A size of total buffer per stream in bytes.
         :param chunk_size: (Default: 4096) A max size of a chunk. Actual chunk may be smaller.
         :param starting_values: (Default: []) An array of strings to put in front of output que.
         """
         starting_values = starting_values or []
         if inputstream:
             input_streamer = StreamFeeder(inputstream)
             input_streamer.start()
             inputstream = input_streamer.output
         # Note: fragile cmd mangling has been removed for use in Kallithea
         assert isinstance(cmd, list), cmd
         _p = subprocess.Popen(cmd, bufsize=-1,
                               stdin=inputstream,
                               stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE,
                               **kwargs)
         bg_out = BufferedGenerator(_p.stdout, buffer_size, chunk_size,
                                    starting_values)
         bg_err = BufferedGenerator(_p.stderr, 16000, 1, bottomless=True)
         while not bg_out.done_reading and not bg_out.reading_paused:
             # doing this until we reach either end of file, or end of buffer.
             bg_out.data_added_event.wait(1)
             bg_out.data_added_event.clear()
         # at this point it's still ambiguous if we are done reading or just full buffer.
         # Either way, if error (returned by ended process, or implied based on
         # presence of stuff in stderr output) we error out.
         # Else, we are happy.
         returncode = _p.poll()
         if (returncode is not None # process has terminated
             and returncode != 0
         ): # and it failed
             bg_out.stop()
             out = b''.join(bg_out)
             bg_err.stop()
             err = b''.join(bg_err)
             if (err.strip() == b'fatal: The remote end hung up unexpectedly' and
                 out.startswith(b'0034shallow ')
             ):
                 # hack inspired by https://github.com/schacon/grack/pull/7
                 bg_out = iter([out])
                 _p = None
             elif err:
                 raise EnvironmentError("Subprocess exited due to an error: %s" % err)
             else:
                 raise EnvironmentError(
                     "Subprocess exited with non 0 ret code: %s" % returncode)
         self.process = _p
         self.output = bg_out
         self.error = bg_err
         self.inputstream = inputstream
     def __iter__(self):
         return self
     def __next__(self):
         if self.process:
             returncode = self.process.poll()
             if (returncode is not None # process has terminated
                 and returncode != 0
             ): # and it failed
                 self.output.stop()
                 self.error.stop()
                 err = ''.join(self.error)
                 raise EnvironmentError("Subprocess exited due to an error:\n" + err)
         return next(self.output)
     def throw(self, type, value=None, traceback=None):
         if self.output.length or not self.output.done_reading:
             raise type(value)
     def close(self):
         try:
             self.process.terminate()
         except:
             pass
         try:
             self.output.close()
         except:
             pass
         try:
             self.error.close()
         except:
             pass
         try:
             os.close(self.inputstream)
         except:
             pass

kallithea/model/db.py

➞

Show inline comments

@@ @@ -16,385 +16,385 @@ kallithea.model.db @@
 ~~~~~~~~~~~~~~~~~~
 Database Models 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 08, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import base64
 import collections
 import datetime
 import functools
 import hashlib
 import logging
 import os
 import time
 import traceback
 import ipaddr
 import sqlalchemy
 from beaker.cache import cache_region, region_invalidate
 from sqlalchemy import Boolean, Column, DateTime, Float, ForeignKey, Index, Integer, LargeBinary, String, Unicode, UnicodeText, UniqueConstraint
 from sqlalchemy.ext.hybrid import hybrid_property
 from sqlalchemy.orm import class_mapper, joinedload, relationship, validates
 from tg.i18n import lazy_ugettext as _
 from webob.exc import HTTPNotFound
 import kallithea
 from kallithea.lib import ext_json
 from kallithea.lib.caching_query import FromCache
 from kallithea.lib.exceptions import DefaultUserException
 from kallithea.lib.utils2 import (
     Optional, ascii_bytes, aslist, get_changeset_safe, get_clone_url, remove_prefix, safe_bytes, safe_int, safe_str, str2bool, urlreadable)
 from kallithea.lib.vcs import get_backend
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.lib.vcs.utils.helpers import get_scm
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 from kallithea.model.meta import Base, Session
 URL_SEP = '/'
 log = logging.getLogger(__name__)
 #==============================================================================
 # BASE CLASSES
 #==============================================================================
 def _hash_key(k):
     return hashlib.md5(safe_bytes(k)).hexdigest()
 class BaseDbModel(object):
     """
     Base Model for all classes
     """
     @classmethod
     def _get_keys(cls):
         """return column names for this model """
         # Note: not a normal dict - iterator gives "users.firstname", but keys gives "firstname"
         return class_mapper(cls).c.keys()
     def get_dict(self):
         """
         return dict with keys and values corresponding
         to this model data """
         d = {}
         for k in self._get_keys():
             d[k] = getattr(self, k)
         # also use __json__() if present to get additional fields
         _json_attr = getattr(self, '__json__', None)
         if _json_attr:
             # update with attributes from __json__
             if callable(_json_attr):
                 _json_attr = _json_attr()
             for k, val in _json_attr.items():
                 d[k] = val
         return d
     def get_appstruct(self):
         """return list with keys and values tuples corresponding
         to this model data """
         return [
             (k, getattr(self, k))
             for k in self._get_keys()
+        ]
     def populate_obj(self, populate_dict):
         """populate model with data from given populate_dict"""
         for k in self._get_keys():
             if k in populate_dict:
                 setattr(self, k, populate_dict[k])
     @classmethod
     def query(cls):
         return Session().query(cls)
     @classmethod
     def get(cls, id_):
         if id_:
             return cls.query().get(id_)
     @classmethod
     def guess_instance(cls, value, callback=None):
         """Haphazardly attempt to convert `value` to a `cls` instance.
         If `value` is None or already a `cls` instance, return it. If `value`
         is a number (or looks like one if you squint just right), assume it's
         a database primary key and let SQLAlchemy sort things out. Otherwise,
         fall back to resolving it using `callback` (if specified); this could
         e.g. be a function that looks up instances by name (though that won't
         work if the name begins with a digit). Otherwise, raise Exception.
         """
         if value is None:
             return None
         if isinstance(value, cls):
             return value
         if isinstance(value, int):
             return cls.get(value)
         if isinstance(value, str) and value.isdigit():
             return cls.get(int(value))
         if callback is not None:
             return callback(value)
         raise Exception(
             'given object must be int, long or Instance of %s '
             'got %s, no callback provided' % (cls, type(value))
+        )
     @classmethod
     def get_or_404(cls, id_):
         try:
             id_ = int(id_)
         except (TypeError, ValueError):
             raise HTTPNotFound
         res = cls.query().get(id_)
         if res is None:
             raise HTTPNotFound
         return res
     @classmethod
     def delete(cls, id_):
         obj = cls.query().get(id_)
         Session().delete(obj)
     def __repr__(self):
         return '<DB:%s>' % (self.__class__.__name__)
 _table_args_default_dict = {'extend_existing': True,
                             'mysql_engine': 'InnoDB',
                             'mysql_charset': 'utf8',
                             'sqlite_autoincrement': True,
+                           }
 class Setting(Base, BaseDbModel):
     __tablename__ = 'settings'
     __table_args__ = (
         _table_args_default_dict,
+    )
     SETTINGS_TYPES = {
         'str': safe_bytes,
         'int': safe_int,
         'unicode': safe_str,
         'bool': str2bool,
         'list': functools.partial(aslist, sep=',')
+    }
     DEFAULT_UPDATE_URL = ''
     app_settings_id = Column(Integer(), primary_key=True)
     app_settings_name = Column(String(255), nullable=False, unique=True)
     _app_settings_value = Column("app_settings_value", Unicode(4096), nullable=False)
     _app_settings_type = Column("app_settings_type", String(255), nullable=True) # FIXME: not nullable?
     def __init__(self, key='', val='', type='unicode'):
         self.app_settings_name = key
         self.app_settings_value = val
         self.app_settings_type = type
     @validates('_app_settings_value')
     def validate_settings_value(self, key, val):
-        assert isinstance(val, unicode)
+        assert isinstance(val, str)
         return val
     @hybrid_property
     def app_settings_value(self):
         v = self._app_settings_value
         _type = self.app_settings_type
         converter = self.SETTINGS_TYPES.get(_type) or self.SETTINGS_TYPES['unicode']
         return converter(v)
     @app_settings_value.setter
     def app_settings_value(self, val):
         """
         Setter that will always make sure we use str in app_settings_value
         """
         self._app_settings_value = safe_str(val)
     @hybrid_property
     def app_settings_type(self):
         return self._app_settings_type
     @app_settings_type.setter
     def app_settings_type(self, val):
         if val not in self.SETTINGS_TYPES:
             raise Exception('type must be one of %s got %s'
                             % (list(self.SETTINGS_TYPES), val))
         self._app_settings_type = val
     def __repr__(self):
         return "<%s %s.%s=%r>" % (
             self.__class__.__name__,
             self.app_settings_name, self.app_settings_type, self.app_settings_value
+        )
     @classmethod
     def get_by_name(cls, key):
         return cls.query() \
             .filter(cls.app_settings_name == key).scalar()
     @classmethod
     def get_by_name_or_create(cls, key, val='', type='unicode'):
         res = cls.get_by_name(key)
         if res is None:
             res = cls(key, val, type)
         return res
     @classmethod
     def create_or_update(cls, key, val=Optional(''), type=Optional('unicode')):
         """
         Creates or updates Kallithea setting. If updates are triggered, it will only
         update parameters that are explicitly set. Optional instance will be skipped.
         :param key:
         :param val:
         :param type:
         :return:
         """
         res = cls.get_by_name(key)
         if res is None:
             val = Optional.extract(val)
             type = Optional.extract(type)
             res = cls(key, val, type)
             Session().add(res)
         else:
             res.app_settings_name = key
             if not isinstance(val, Optional):
                 # update if set
                 res.app_settings_value = val
             if not isinstance(type, Optional):
                 # update if set
                 res.app_settings_type = type
         return res
     @classmethod
     def get_app_settings(cls, cache=False):
         ret = cls.query()
         if cache:
             ret = ret.options(FromCache("sql_cache_short", "get_hg_settings"))
         if ret is None:
             raise Exception('Could not get application settings !')
         settings = {}
         for each in ret:
             settings[each.app_settings_name] = \
                 each.app_settings_value
         return settings
     @classmethod
     def get_auth_settings(cls, cache=False):
         ret = cls.query() \
                 .filter(cls.app_settings_name.startswith('auth_')).all()
         fd = {}
         for row in ret:
             fd[row.app_settings_name] = row.app_settings_value
         return fd
     @classmethod
     def get_default_repo_settings(cls, cache=False, strip_prefix=False):
         ret = cls.query() \
                 .filter(cls.app_settings_name.startswith('default_')).all()
         fd = {}
         for row in ret:
             key = row.app_settings_name
             if strip_prefix:
                 key = remove_prefix(key, prefix='default_')
             fd.update({key: row.app_settings_value})
         return fd
     @classmethod
     def get_server_info(cls):
         import pkg_resources
         import platform
         from kallithea.lib.utils import check_git_version
         mods = [(p.project_name, p.version) for p in pkg_resources.working_set]
         info = {
             'modules': sorted(mods, key=lambda k: k[0].lower()),
             'py_version': platform.python_version(),
             'platform': platform.platform(),
             'kallithea_version': kallithea.__version__,
             'git_version': str(check_git_version()),
             'git_path': kallithea.CONFIG.get('git_path')
+        }
         return info
 class Ui(Base, BaseDbModel):
     __tablename__ = 'ui'
     __table_args__ = (
         # FIXME: ui_key as key is wrong and should be removed when the corresponding
         # Ui.get_by_key has been replaced by the composite key
         UniqueConstraint('ui_key'),
         UniqueConstraint('ui_section', 'ui_key'),
         _table_args_default_dict,
+    )
     HOOK_UPDATE = 'changegroup.update'
     HOOK_REPO_SIZE = 'changegroup.repo_size'
     ui_id = Column(Integer(), primary_key=True)
     ui_section = Column(String(255), nullable=False)
     ui_key = Column(String(255), nullable=False)
     ui_value = Column(String(255), nullable=True) # FIXME: not nullable?
     ui_active = Column(Boolean(), nullable=False, default=True)
     @classmethod
     def get_by_key(cls, section, key):
         """ Return specified Ui object, or None if not found. """
         return cls.query().filter_by(ui_section=section, ui_key=key).scalar()
     @classmethod
     def get_or_create(cls, section, key):
         """ Return specified Ui object, creating it if necessary. """
         setting = cls.get_by_key(section, key)
         if setting is None:
             setting = cls(ui_section=section, ui_key=key)
             Session().add(setting)
         return setting
     @classmethod
     def get_builtin_hooks(cls):
         q = cls.query()
         q = q.filter(cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE]))
         q = q.filter(cls.ui_section == 'hooks')
         return q.all()
     @classmethod
     def get_custom_hooks(cls):
         q = cls.query()
         q = q.filter(~cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE]))
         q = q.filter(cls.ui_section == 'hooks')
         return q.all()
     @classmethod
     def get_repos_location(cls):
         return cls.get_by_key('paths', '/').ui_value
     @classmethod
     def create_or_update_hook(cls, key, val):
         new_ui = cls.get_or_create('hooks', key)
         new_ui.ui_active = True
         new_ui.ui_value = val
     def __repr__(self):
         return '<%s %s.%s=%r>' % (
             self.__class__.__name__,
             self.ui_section, self.ui_key, self.ui_value)
 class User(Base, BaseDbModel):

kallithea/model/gist.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.gist
 ~~~~~~~~~~~~~~~~~~~~
 gist 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: May 9, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import os
 import random
 import shutil
 import time
 import traceback
 from kallithea.lib import ext_json
 from kallithea.lib.utils2 import AttributeDict, ascii_bytes, safe_int, time_to_datetime
 from kallithea.model.db import Gist, Session, User
 from kallithea.model.repo import RepoModel
 from kallithea.model.scm import ScmModel
 log = logging.getLogger(__name__)
 GIST_STORE_LOC = '.rc_gist_store'
 GIST_METADATA_FILE = '.rc_gist_metadata'
 def make_gist_access_id():
     """Generate a random, URL safe, almost certainly unique gist identifier."""
     rnd = random.SystemRandom() # use cryptographically secure system PRNG
     alphabet = '23456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghjklmnpqrstuvwxyz'
     length = 20
     return u''.join(rnd.choice(alphabet) for _ in range(length))
 class GistModel(object):
     def __delete_gist(self, gist):
         """
         removes gist from filesystem
         :param gist: gist object
         """
         root_path = RepoModel().repos_path
         rm_path = os.path.join(root_path, GIST_STORE_LOC, gist.gist_access_id)
         log.info("Removing %s", rm_path)
         shutil.rmtree(rm_path)
     def _store_metadata(self, repo, gist_id, gist_access_id, user_id, gist_type,
                         gist_expires):
         """
         store metadata inside the gist, this can be later used for imports
         or gist identification
         """
         metadata = {
             'metadata_version': '1',
             'gist_db_id': gist_id,
             'gist_access_id': gist_access_id,
             'gist_owner_id': user_id,
             'gist_type': gist_type,
             'gist_expires': gist_expires,
             'gist_updated': time.time(),
+        }
         with open(os.path.join(repo.path, '.hg', GIST_METADATA_FILE), 'wb') as f:
             f.write(ascii_bytes(ext_json.dumps(metadata)))
     def get_gist(self, gist):
         return Gist.guess_instance(gist)
     def get_gist_files(self, gist_access_id, revision=None):
         """
         Get files for given gist
         :param gist_access_id:
         """
         repo = Gist.get_by_access_id(gist_access_id)
         cs = repo.scm_instance.get_changeset(revision)
         return cs, [n for n in cs.get_node('/')]
     def create(self, description, owner, ip_addr, gist_mapping,
                gist_type=Gist.GIST_PUBLIC, lifetime=-1):
         """
         :param description: description of the gist
         :param owner: user who created this gist
         :param gist_mapping: mapping {filename:{'content':content},...}
         :param gist_type: type of gist private/public
         :param lifetime: in minutes, -1 == forever
         """
         owner = User.guess_instance(owner)
         gist_access_id = make_gist_access_id()
         lifetime = safe_int(lifetime, -1)
         gist_expires = time.time() + (lifetime * 60) if lifetime != -1 else -1
         log.debug('set GIST expiration date to: %s',
                   time_to_datetime(gist_expires)
                    if gist_expires != -1 else 'forever')
         # create the Database version
         gist = Gist()
         gist.gist_description = description
         gist.gist_access_id = gist_access_id
         gist.owner_id = owner.user_id
         gist.gist_expires = gist_expires
         gist.gist_type = gist_type
         Session().add(gist)
         Session().flush() # make database assign gist.gist_id
         if gist_type == Gist.GIST_PUBLIC:
             # use DB ID for easy to use GIST ID
-            gist.gist_access_id = unicode(gist.gist_id)
+            gist.gist_access_id = str(gist.gist_id)
         log.debug('Creating new %s GIST repo %s', gist_type, gist.gist_access_id)
         repo = RepoModel()._create_filesystem_repo(
             repo_name=gist.gist_access_id, repo_type='hg', repo_group=GIST_STORE_LOC)
         processed_mapping = {}
         for filename in gist_mapping:
             if filename != os.path.basename(filename):
                 raise Exception('Filename cannot be inside a directory')
             content = gist_mapping[filename]['content']
             # TODO: expand support for setting explicit lexers
 #             if lexer is None:
 #                 try:
 #                     guess_lexer = pygments.lexers.guess_lexer_for_filename
 #                     lexer = guess_lexer(filename,content)
 #                 except pygments.util.ClassNotFound:
 #                     lexer = 'text'
             processed_mapping[filename] = {'content': content}
         # now create single multifile commit
         message = 'added file'
         message += 's: ' if len(processed_mapping) > 1 else ': '
         message += ', '.join([x for x in processed_mapping])
         # fake Kallithea Repository object
         fake_repo = AttributeDict(dict(
             repo_name=os.path.join(GIST_STORE_LOC, gist.gist_access_id),
             scm_instance_no_cache=lambda: repo,
         ))
         ScmModel().create_nodes(
             user=owner.user_id,
             ip_addr=ip_addr,
             repo=fake_repo,
             message=message,
             nodes=processed_mapping,
             trigger_push_hook=False
+        )
         self._store_metadata(repo, gist.gist_id, gist.gist_access_id,
                              owner.user_id, gist.gist_type, gist.gist_expires)
         return gist
     def delete(self, gist, fs_remove=True):
         gist = Gist.guess_instance(gist)
         try:
             Session().delete(gist)
             if fs_remove:
                 self.__delete_gist(gist)
             else:
                 log.debug('skipping removal from filesystem')
         except Exception:
             log.error(traceback.format_exc())
             raise
     def update(self, gist, description, owner, ip_addr, gist_mapping, gist_type,
                lifetime):
         gist = Gist.guess_instance(gist)
         gist_repo = gist.scm_instance
         lifetime = safe_int(lifetime, -1)
         if lifetime == 0:  # preserve old value
             gist_expires = gist.gist_expires
         else:
             gist_expires = time.time() + (lifetime * 60) if lifetime != -1 else -1
         # calculate operation type based on given data
         gist_mapping_op = {}
         for k, v in gist_mapping.items():
             # add, mod, del
             if not v['org_filename'] and v['filename']:
                 op = 'add'
             elif v['org_filename'] and not v['filename']:
                 op = 'del'
             else:
                 op = 'mod'
             v['op'] = op
             gist_mapping_op[k] = v
         gist.gist_description = description
         gist.gist_expires = gist_expires
         gist.owner = owner
         gist.gist_type = gist_type
         message = 'updated file'
         message += 's: ' if len(gist_mapping) > 1 else ': '
         message += ', '.join([x for x in gist_mapping])
         # fake Kallithea Repository object
         fake_repo = AttributeDict(dict(
             repo_name=os.path.join(GIST_STORE_LOC, gist.gist_access_id),
             scm_instance_no_cache=lambda: gist_repo,
         ))
         self._store_metadata(gist_repo, gist.gist_id, gist.gist_access_id,
                              owner.user_id, gist.gist_type, gist.gist_expires)
         ScmModel().update_nodes(
             user=owner.user_id,
             ip_addr=ip_addr,
             repo=fake_repo,
             message=message,
             nodes=gist_mapping_op,
             trigger_push_hook=False
+        )
         return gist

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 _
 import kallithea
 from kallithea.lib import helpers as h
 from kallithea.model.db import User
 log = logging.getLogger(__name__)
 class NotificationModel(object):
     TYPE_CHANGESET_COMMENT = u'cs_comment'
     TYPE_MESSAGE = u'message'
     TYPE_MENTION = u'mention' # not used
     TYPE_REGISTRATION = u'registration'
     TYPE_PULL_REQUEST = u'pull_request'
     TYPE_PULL_REQUEST_COMMENT = u'pull_request_comment'
     def create(self, created_by, subject, 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 subject:
         :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
         """
         from kallithea.lib.celerylib import tasks
         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 = User.guess_instance(created_by)
         recipients_objs = set()
         if recipients:
             for u in recipients:
                 obj = 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 = User.query().filter(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 = h.fmt_date(datetime.datetime.now())
         html_kwargs = {
                   'subject': subject,
                   'body': h.render_w_mentions(body, repo_name),
                   'when': created_on,
                   'user': created_by_obj.username,
+                  }
         txt_kwargs = {
                   'subject': subject,
                   '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 person who created this comment
         rec_objs = set(recipients_objs).difference(set([created_by_obj]))
         # send email with notification to all other participants
         for rec in rec_objs:
             tasks.send_email([rec.email], email_subject, email_txt_body,
                      email_html_body, headers, author=created_by_obj)
 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._template_root = kallithea.CONFIG['paths']['templates'][0]
         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'),
             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(unicode(status_change))  # apply unicode to evaluate LazyString before .join
+            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):
         """
         return generated template for email based on given type
         """
         base = 'email_templates/' + self.email_types.get(type_, self.email_types[self.TYPE_DEFAULT]) + '.' + content_type
         email_template = self._tmpl_lookup.get_template(base)
         # translator and helpers inject
         _kwargs = {'_': _,
                    'h': h,
                    'c': c}
         _kwargs.update(kwargs)
         if content_type == 'html':
             _kwargs.update({
                 "color_text": "#202020",
                 "color_emph": "#395fa0",
                 "color_link": "#395fa0",
                 "color_border": "#ddd",
                 "color_background_grey": "#f9f9f9",
                 "color_button": "#395fa0",
                 "monospace_style": "font-family:Lucida Console,Consolas,Monaco,Inconsolata,Liberation Mono,monospace",
                 "sans_style": "font-family:Helvetica,Arial,sans-serif",
                 })
             _kwargs.update({
                 "default_style": "%(sans_style)s;font-weight:200;font-size:14px;line-height:17px;color:%(color_text)s" % _kwargs,
                 "comment_style": "%(monospace_style)s;white-space:pre-wrap" % _kwargs,
                 "data_style": "border:%(color_border)s 1px solid;background:%(color_background_grey)s" % _kwargs,
                 "emph_style": "font-weight:600;color:%(color_emph)s" % _kwargs,
                 "link_style": "color:%(color_link)s;text-decoration:none" % _kwargs,
                 "link_text_style": "color:%(color_text)s;text-decoration:none;border:%(color_border)s 1px solid;background:%(color_background_grey)s" % _kwargs,
                 })
         log.debug('rendering tmpl %s with kwargs %s', base, _kwargs)
         return email_template.render_unicode(**_kwargs)

kallithea/tests/vcs/test_git.py

➞

Show inline comments

@@ @@ -399,391 +399,391 @@ class TestGitChangeset(object): @@
                 '0115510b70c7229dbc5dc49036b32e7d91d23acd',
                 '7cb3fd1b6d8c20ba89e2264f1c8baebc8a52d36e',
                 '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                 '191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             ],
             'vcs/nodes.py': [
                 '33fa3223355104431402a888fa77a4e9956feb3e',
                 'fa014c12c26d10ba682fadb78f2a11c24c8118e1',
                 'e686b958768ee96af8029fe19c6050b1a8dd3b2b',
                 'ab5721ca0a081f26bf43d9051e615af2cc99952f',
                 'c877b68d18e792a66b7f4c529ea02c8f80801542',
                 '4313566d2e417cb382948f8d9d7c765330356054',
                 '6c2303a793671e807d1cfc70134c9ca0767d98c2',
                 '54386793436c938cff89326944d4c2702340037d',
                 '54000345d2e78b03a99d561399e8e548de3f3203',
                 '1c6b3677b37ea064cb4b51714d8f7498f93f4b2b',
                 '2d03ca750a44440fb5ea8b751176d1f36f8e8f46',
                 '2a08b128c206db48c2f0b8f70df060e6db0ae4f8',
                 '30c26513ff1eb8e5ce0e1c6b477ee5dc50e2f34b',
                 'ac71e9503c2ca95542839af0ce7b64011b72ea7c',
                 '12669288fd13adba2a9b7dd5b870cc23ffab92d2',
                 '5a0c84f3e6fe3473e4c8427199d5a6fc71a9b382',
                 '12f2f5e2b38e6ff3fbdb5d722efed9aa72ecb0d5',
                 '5eab1222a7cd4bfcbabc218ca6d04276d4e27378',
                 'f50f42baeed5af6518ef4b0cb2f1423f3851a941',
                 'd7e390a45f6aa96f04f5e7f583ad4f867431aa25',
                 'f15c21f97864b4f071cddfbf2750ec2e23859414',
                 'e906ef056cf539a4e4e5fc8003eaf7cf14dd8ade',
                 'ea2b108b48aa8f8c9c4a941f66c1a03315ca1c3b',
                 '84dec09632a4458f79f50ddbbd155506c460b4f9',
                 '0115510b70c7229dbc5dc49036b32e7d91d23acd',
                 '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                 '3bf1c5868e570e39569d094f922d33ced2fa3b2b',
                 'b8d04012574729d2c29886e53b1a43ef16dd00a1',
                 '6970b057cffe4aab0a792aa634c89f4bebf01441',
                 'dd80b0f6cf5052f17cc738c2951c4f2070200d7f',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             ],
             'vcs/backends/git.py': [
                 '4cf116ad5a457530381135e2f4c453e68a1b0105',
                 '9a751d84d8e9408e736329767387f41b36935153',
                 'cb681fb539c3faaedbcdf5ca71ca413425c18f01',
                 '428f81bb652bcba8d631bce926e8834ff49bdcc6',
                 '180ab15aebf26f98f714d8c68715e0f05fa6e1c7',
                 '2b8e07312a2e89e92b90426ab97f349f4bce2a3a',
                 '50e08c506174d8645a4bb517dd122ac946a0f3bf',
                 '54000345d2e78b03a99d561399e8e548de3f3203',
             ],
+        }
         for path, revs in files.items():
             node = self.repo.get_changeset(revs[0]).get_node(path)
             node_revs = [chset.raw_id for chset in node.history]
             assert set(revs).issubset(set(node_revs)), "We assumed that %s is subset of revisions for which file %s " \
                 "has been changed, and history of that node returned: %s" \
                 % (revs, path, node_revs)
     def test_file_annotate(self):
         files = {
             'vcs/backends/__init__.py': {
                 'c1214f7e79e02fc37156ff215cd71275450cffc3': {
                     'lines_no': 1,
                     'changesets': [
                         'c1214f7e79e02fc37156ff215cd71275450cffc3',
                     ],
                 },
                 '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647': {
                     'lines_no': 21,
                     'changesets': [
                         '49d3fd156b6f7db46313fac355dca1a0b94a0017',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                     ],
                 },
                 'e29b67bd158580fc90fc5e9111240b90e6e86064': {
                     'lines_no': 32,
                     'changesets': [
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '5eab1222a7cd4bfcbabc218ca6d04276d4e27378',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '54000345d2e78b03a99d561399e8e548de3f3203',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '78c3f0c23b7ee935ec276acb8b8212444c33c396',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '78c3f0c23b7ee935ec276acb8b8212444c33c396',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                     ],
                 },
             },
+        }
         for fname, revision_dict in files.items():
             for rev, data in revision_dict.items():
                 cs = self.repo.get_changeset(rev)
                 l1_1 = [x[1] for x in cs.get_file_annotate(fname)]
                 l1_2 = [x[2]().raw_id for x in cs.get_file_annotate(fname)]
                 assert l1_1 == l1_2
                 l1 = l1_1
                 l2 = files[fname][rev]['changesets']
                 assert l1 == l2, "The lists of revision for %s@rev %s" \
                     "from annotation list should match each other, " \
                     "got \n%s \nvs \n%s " % (fname, rev, l1, l2)
     def test_files_state(self):
         """
         Tests state of FileNodes.
         """
         node = self.repo \
             .get_changeset('e6ea6d16e2f26250124a1f4b4fe37a912f9d86a0') \
             .get_node('vcs/utils/diffs.py')
         assert node.state, NodeState.ADDED
         assert node.added
         assert not node.changed
         assert not node.not_changed
         assert not node.removed
         node = self.repo \
             .get_changeset('33fa3223355104431402a888fa77a4e9956feb3e') \
             .get_node('.hgignore')
         assert node.state, NodeState.CHANGED
         assert not node.added
         assert node.changed
         assert not node.not_changed
         assert not node.removed
         node = self.repo \
             .get_changeset('e29b67bd158580fc90fc5e9111240b90e6e86064') \
             .get_node('setup.py')
         assert node.state, NodeState.NOT_CHANGED
         assert not node.added
         assert not node.changed
         assert node.not_changed
         assert not node.removed
         # If node has REMOVED state then trying to fetch it would raise
         # ChangesetError exception
         chset = self.repo.get_changeset(
             'fa6600f6848800641328adbf7811fd2372c02ab2')
         path = 'vcs/backends/BaseRepository.py'
         with pytest.raises(NodeDoesNotExistError):
             chset.get_node(path)
         # but it would be one of ``removed`` (changeset's attribute)
         assert path in [rf.path for rf in chset.removed]
         chset = self.repo.get_changeset(
             '54386793436c938cff89326944d4c2702340037d')
         changed = ['setup.py', 'tests/test_nodes.py', 'vcs/backends/hg.py',
             'vcs/nodes.py']
         assert set(changed) == set([f.path for f in chset.changed])
-    def test_commit_message_is_unicode(self):
+    def test_commit_message_is_str(self):
         for cs in self.repo:
-            assert isinstance(cs.message, unicode)
+            assert isinstance(cs.message, str)
-    def test_changeset_author_is_unicode(self):
+    def test_changeset_author_is_str(self):
         for cs in self.repo:
-            assert isinstance(cs.author, unicode)
+            assert isinstance(cs.author, str)
     def test_repo_files_content_is_bytes(self):
         changeset = self.repo.get_changeset()
         for node in changeset.get_node('/'):
             if node.is_file():
                 assert isinstance(node.content, bytes)
     def test_wrong_path(self):
         # There is 'setup.py' in the root dir but not there:
         path = 'foo/bar/setup.py'
         tip = self.repo.get_changeset()
         with pytest.raises(VCSError):
             tip.get_node(path)
     def test_author_email(self):
         assert 'marcin@python-blog.com' == self.repo.get_changeset('c1214f7e79e02fc37156ff215cd71275450cffc3').author_email
         assert 'lukasz.balcerzak@python-center.pl' == self.repo.get_changeset('ff7ca51e58c505fec0dd2491de52c622bb7a806b').author_email
         assert '' == self.repo.get_changeset('8430a588b43b5d6da365400117c89400326e7992').author_email
     def test_author_username(self):
         assert 'Marcin Kuzminski' == self.repo.get_changeset('c1214f7e79e02fc37156ff215cd71275450cffc3').author_name
         assert 'Lukasz Balcerzak' == self.repo.get_changeset('ff7ca51e58c505fec0dd2491de52c622bb7a806b').author_name
         assert 'marcink none@none' == self.repo.get_changeset('8430a588b43b5d6da365400117c89400326e7992').author_name
 class TestGitSpecificWithRepo(_BackendTestMixin):
     backend_alias = 'git'
     @classmethod
     def _get_commits(cls):
         return [
+            {
                 'message': 'Initial',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('foobar/static/js/admin/base.js', content='base'),
                     FileNode('foobar/static/admin', content='admin',
                         mode=0o120000), # this is a link
                     FileNode('foo', content='foo'),
                 ],
             },
+            {
                 'message': 'Second',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'added': [
                     FileNode('foo2', content='foo2'),
                 ],
             },
+        ]
     def test_paths_slow_traversing(self):
         cs = self.repo.get_changeset()
         assert cs.get_node('foobar').get_node('static').get_node('js').get_node('admin').get_node('base.js').content == b'base'
     def test_paths_fast_traversing(self):
         cs = self.repo.get_changeset()
         assert cs.get_node('foobar/static/js/admin/base.js').content == b'base'
     def test_workdir_get_branch(self):
         self.repo.run_git_command(['checkout', '-b', 'production'])
         # Regression test: one of following would fail if we don't check
         # .git/HEAD file
         self.repo.run_git_command(['checkout', 'production'])
         assert self.repo.workdir.get_branch() == 'production'
         self.repo.run_git_command(['checkout', 'master'])
         assert self.repo.workdir.get_branch() == 'master'
     def test_get_diff_runs_git_command_with_hashes(self):
         self.repo._run_git_command = mock.Mock(return_value=(b'', b''))
         self.repo.get_diff(0, 1)
         self.repo._run_git_command.assert_called_once_with(
             ['diff', '-U3', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1)], cwd=self.repo.path)
     def test_get_diff_runs_git_command_with_str_hashes(self):
         self.repo._run_git_command = mock.Mock(return_value=(b'', b''))
         self.repo.get_diff(self.repo.EMPTY_CHANGESET, 1)
         self.repo._run_git_command.assert_called_once_with(
             ['show', '-U3', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(1)], cwd=self.repo.path)
     def test_get_diff_runs_git_command_with_path_if_its_given(self):
         self.repo._run_git_command = mock.Mock(return_value=(b'', b''))
         self.repo.get_diff(0, 1, 'foo')
         self.repo._run_git_command.assert_called_once_with(
             ['diff', '-U3', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'], cwd=self.repo.path)
     def test_get_diff_does_not_sanitize_valid_context(self):
         almost_overflowed_long_int = 2**31-1
         self.repo._run_git_command = mock.Mock(return_value=(b'', b''))
         self.repo.get_diff(0, 1, 'foo', context=almost_overflowed_long_int)
         self.repo._run_git_command.assert_called_once_with(
             ['diff', '-U' + str(almost_overflowed_long_int), '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'], cwd=self.repo.path)
     def test_get_diff_sanitizes_overflowing_context(self):
         overflowed_long_int = 2**31
         sanitized_overflowed_long_int = overflowed_long_int-1
         self.repo._run_git_command = mock.Mock(return_value=(b'', b''))
         self.repo.get_diff(0, 1, 'foo', context=overflowed_long_int)
         self.repo._run_git_command.assert_called_once_with(
             ['diff', '-U' + str(sanitized_overflowed_long_int), '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'], cwd=self.repo.path)
     def test_get_diff_does_not_sanitize_zero_context(self):
         zero_context = 0
         self.repo._run_git_command = mock.Mock(return_value=(b'', b''))
         self.repo.get_diff(0, 1, 'foo', context=zero_context)
         self.repo._run_git_command.assert_called_once_with(
             ['diff', '-U' + str(zero_context), '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'], cwd=self.repo.path)
     def test_get_diff_sanitizes_negative_context(self):
         negative_context = -10
         self.repo._run_git_command = mock.Mock(return_value=(b'', b''))
         self.repo.get_diff(0, 1, 'foo', context=negative_context)
         self.repo._run_git_command.assert_called_once_with(
             ['diff', '-U0', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'], cwd=self.repo.path)
 class TestGitRegression(_BackendTestMixin):
     backend_alias = 'git'
     @classmethod
     def _get_commits(cls):
         return [
+            {
                 'message': 'Initial',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('bot/__init__.py', content='base'),
                     FileNode('bot/templates/404.html', content='base'),
                     FileNode('bot/templates/500.html', content='base'),
                 ],
             },
+            {
                 'message': 'Second',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'added': [
                     FileNode('bot/build/migrations/1.py', content='foo2'),
                     FileNode('bot/build/migrations/2.py', content='foo2'),
                     FileNode('bot/build/static/templates/f.html', content='foo2'),
                     FileNode('bot/build/static/templates/f1.html', content='foo2'),
                     FileNode('bot/build/templates/err.html', content='foo2'),
                     FileNode('bot/build/templates/err2.html', content='foo2'),
                 ],
             },
+        ]
     def test_similar_paths(self):
         cs = self.repo.get_changeset()
         paths = lambda *n: [x.path for x in n]
         assert paths(*cs.get_nodes('bot')) == ['bot/build', 'bot/templates', 'bot/__init__.py']
         assert paths(*cs.get_nodes('bot/build')) == ['bot/build/migrations', 'bot/build/static', 'bot/build/templates']
         assert paths(*cs.get_nodes('bot/build/static')) == ['bot/build/static/templates']
         # this get_nodes below causes troubles !
         assert paths(*cs.get_nodes('bot/build/static/templates')) == ['bot/build/static/templates/f.html', 'bot/build/static/templates/f1.html']
         assert paths(*cs.get_nodes('bot/build/templates')) == ['bot/build/templates/err.html', 'bot/build/templates/err2.html']
         assert paths(*cs.get_nodes('bot/templates/')) == ['bot/templates/404.html', 'bot/templates/500.html']
 class TestGitHooks(object):
     """
     Tests related to hook functionality of Git repositories.
     """
     def setup_method(self):
         # For each run we want a fresh repo.
         self.repo_directory = get_new_dir("githookrepo")
         self.repo = GitRepository(self.repo_directory, create=True)
         # Create a dictionary where keys are hook names, and values are paths to
         # them in the non-bare repo. Deduplicates code in tests a bit.
         self.kallithea_hooks = {
             "pre-receive": os.path.join(self.repo.path, '.git', 'hooks', "pre-receive"),
             "post-receive": os.path.join(self.repo.path, '.git', 'hooks', "post-receive"),
+        }
     def test_hooks_created_if_missing(self):

kallithea/tests/vcs/test_hg.py

➞

Show inline comments

@@ @@ -346,246 +346,246 @@ class TestMercurialChangeset(object): @@
             (10, 'vcs/backends/base.py', 2921),
             (20, 'vcs/backends/base.py', 3936),
             (60, 'vcs/backends/base.py', 6189),
+        )
         for revision, path, size in to_check:
             self._test_file_size(revision, path, size)
     def _test_dir_size(self, revision, path, size):
         node = self.repo.get_changeset(revision).get_node(path)
         assert not node.is_file()
         assert node.size == size
     def test_dir_size(self):
         to_check = (
             ('96507bd11ecc', '/', 682421),
             ('a53d9201d4bc', '/', 682410),
             ('90243de06161', '/', 682006),
+        )
         for revision, path, size in to_check:
             self._test_dir_size(revision, path, size)
     def test_repo_size(self):
         assert self.repo.size == 682421
     def test_file_history(self):
         # we can only check if those revisions are present in the history
         # as we cannot update this test every time file is changed
         files = {
             'setup.py': [7, 18, 45, 46, 47, 69, 77],
             'vcs/nodes.py': [7, 8, 24, 26, 30, 45, 47, 49, 56, 57, 58, 59, 60,
 , 73, 76],
             'vcs/backends/hg.py': [4, 5, 6, 11, 12, 13, 14, 15, 16, 21, 22, 23,
 , 27, 28, 30, 31, 33, 35, 36, 37, 38, 39, 40, 41, 44, 45, 47,
 , 49, 53, 54, 55, 58, 60, 61, 67, 68, 69, 70, 73, 77, 78, 79,
 ],
+        }
         for path, revs in files.items():
             tip = self.repo.get_changeset(revs[-1])
             node = tip.get_node(path)
             node_revs = [chset.revision for chset in node.history]
             assert set(revs).issubset(set(node_revs)), \
                 "We assumed that %s is subset of revisions for which file %s " \
                 "has been changed, and history of that node returned: %s" \
                 % (revs, path, node_revs)
     def test_file_annotate(self):
         files = {
                  'vcs/backends/__init__.py':
                   {89: {'lines_no': 31,
                         'changesets': [32, 32, 61, 32, 32, 37, 32, 32, 32, 44,
 , 37, 37, 37, 45, 37, 44, 37, 37, 37,
 , 32, 32, 32, 37, 32, 37, 37, 32,
 , 32]},
 : {'lines_no': 1,
                         'changesets': [4]},
 : {'lines_no': 31,
                         'changesets': [32, 32, 45, 32, 32, 37, 32, 32, 32, 44,
 , 37, 37, 37, 45, 37, 44, 37, 37, 37,
 , 32, 32, 32, 37, 32, 37, 37, 32,
 , 32]}},
                  'vcs/exceptions.py':
                  {89: {'lines_no': 18,
                        'changesets': [16, 16, 16, 16, 16, 16, 16, 16, 16, 16,
 , 16, 17, 16, 16, 18, 18, 18]},
 : {'lines_no': 18,
                        'changesets': [16, 16, 16, 16, 16, 16, 16, 16, 16, 16,
 , 16, 17, 16, 16, 18, 18, 18]},
 : {'lines_no': 18, 'changesets': [16, 16, 16, 16, 16, 16,
 , 16, 16, 16, 16, 16,
 , 16, 16, 18, 18, 18]}},
                  'MANIFEST.in': {89: {'lines_no': 5,
                                       'changesets': [7, 7, 7, 71, 71]},
 : {'lines_no': 3,
                                       'changesets': [7, 7, 7]},
 : {'lines_no': 3,
                                      'changesets': [7, 7, 7]}}}
         for fname, revision_dict in files.items():
             for rev, data in revision_dict.items():
                 cs = self.repo.get_changeset(rev)
                 l1_1 = [x[1] for x in cs.get_file_annotate(fname)]
                 l1_2 = [x[2]().raw_id for x in cs.get_file_annotate(fname)]
                 assert l1_1 == l1_2
                 l1 = l1_2 = [x[2]().revision for x in cs.get_file_annotate(fname)]
                 l2 = files[fname][rev]['changesets']
                 assert l1 == l2, "The lists of revision for %s@rev%s" \
                     "from annotation list should match each other," \
                     "got \n%s \nvs \n%s " % (fname, rev, l1, l2)
     def test_changeset_state(self):
         """
         Tests which files have been added/changed/removed at particular revision
         """
         # rev 46ad32a4f974:
         # hg st --rev 46ad32a4f974
         #    changed: 13
         #    added:   20
         #    removed: 1
         changed = set(['.hgignore'
             , 'README.rst', 'docs/conf.py', 'docs/index.rst', 'setup.py'
             , 'tests/test_hg.py', 'tests/test_nodes.py', 'vcs/__init__.py'
             , 'vcs/backends/__init__.py', 'vcs/backends/base.py'
             , 'vcs/backends/hg.py', 'vcs/nodes.py', 'vcs/utils/__init__.py'])
         added = set(['docs/api/backends/hg.rst'
             , 'docs/api/backends/index.rst', 'docs/api/index.rst'
             , 'docs/api/nodes.rst', 'docs/api/web/index.rst'
             , 'docs/api/web/simplevcs.rst', 'docs/installation.rst'
             , 'docs/quickstart.rst', 'setup.cfg', 'vcs/utils/baseui_config.py'
             , 'vcs/utils/web.py', 'vcs/web/__init__.py', 'vcs/web/exceptions.py'
             , 'vcs/web/simplevcs/__init__.py', 'vcs/web/simplevcs/exceptions.py'
             , 'vcs/web/simplevcs/middleware.py', 'vcs/web/simplevcs/models.py'
             , 'vcs/web/simplevcs/settings.py', 'vcs/web/simplevcs/utils.py'
             , 'vcs/web/simplevcs/views.py'])
         removed = set(['docs/api.rst'])
         chset64 = self.repo.get_changeset('46ad32a4f974')
         assert set((node.path for node in chset64.added)) == added
         assert set((node.path for node in chset64.changed)) == changed
         assert set((node.path for node in chset64.removed)) == removed
         # rev b090f22d27d6:
         # hg st --rev b090f22d27d6
         #    changed: 13
         #    added:   20
         #    removed: 1
         chset88 = self.repo.get_changeset('b090f22d27d6')
         assert set((node.path for node in chset88.added)) == set()
         assert set((node.path for node in chset88.changed)) == set(['.hgignore'])
         assert set((node.path for node in chset88.removed)) == set()
         # 85:
         #    added:   2 ['vcs/utils/diffs.py', 'vcs/web/simplevcs/views/diffs.py']
         #    changed: 4 ['vcs/web/simplevcs/models.py', ...]
         #    removed: 1 ['vcs/utils/web.py']
         chset85 = self.repo.get_changeset(85)
         assert set((node.path for node in chset85.added)) == set([
             'vcs/utils/diffs.py',
             'vcs/web/simplevcs/views/diffs.py'
         ])
         assert set((node.path for node in chset85.changed)) == set([
             'vcs/web/simplevcs/models.py',
             'vcs/web/simplevcs/utils.py',
             'vcs/web/simplevcs/views/__init__.py',
             'vcs/web/simplevcs/views/repository.py',
         ])
         assert set((node.path for node in chset85.removed)) == set([
             'vcs/utils/web.py'
         ])
     def test_files_state(self):
         """
         Tests state of FileNodes.
         """
         chset = self.repo.get_changeset(85)
         node = chset.get_node('vcs/utils/diffs.py')
         assert node.state, NodeState.ADDED
         assert node.added
         assert not node.changed
         assert not node.not_changed
         assert not node.removed
         chset = self.repo.get_changeset(88)
         node = chset.get_node('.hgignore')
         assert node.state, NodeState.CHANGED
         assert not node.added
         assert node.changed
         assert not node.not_changed
         assert not node.removed
         chset = self.repo.get_changeset(85)
         node = chset.get_node('setup.py')
         assert node.state, NodeState.NOT_CHANGED
         assert not node.added
         assert not node.changed
         assert node.not_changed
         assert not node.removed
         # If node has REMOVED state then trying to fetch it would raise
         # ChangesetError exception
         chset = self.repo.get_changeset(2)
         path = 'vcs/backends/BaseRepository.py'
         with pytest.raises(NodeDoesNotExistError):
             chset.get_node(path)
         # but it would be one of ``removed`` (changeset's attribute)
         assert path in [rf.path for rf in chset.removed]
-    def test_commit_message_is_unicode(self):
+    def test_commit_message_is_str(self):
         for cm in self.repo:
-            assert isinstance(cm.message, unicode)
+            assert isinstance(cm.message, str)
-    def test_changeset_author_is_unicode(self):
+    def test_changeset_author_is_str(self):
         for cm in self.repo:
-            assert isinstance(cm.author, unicode)
+            assert isinstance(cm.author, str)
     def test_repo_files_content_is_bytes(self):
         test_changeset = self.repo.get_changeset(100)
         for node in test_changeset.get_node('/'):
             if node.is_file():
                 assert isinstance(node.content, bytes)
     def test_wrong_path(self):
         # There is 'setup.py' in the root dir but not there:
         path = 'foo/bar/setup.py'
         with pytest.raises(VCSError):
             self.repo.get_changeset().get_node(path)
     def test_archival_file(self):
         # TODO:
         pass
     def test_archival_as_generator(self):
         # TODO:
         pass
     def test_archival_wrong_kind(self):
         tip = self.repo.get_changeset()
         with pytest.raises(VCSError):
             tip.fill_archive(kind='error')
     def test_archival_empty_prefix(self):
         # TODO:
         pass
     def test_author_email(self):
         assert 'marcin@python-blog.com' == self.repo.get_changeset('b986218ba1c9').author_email
         assert 'lukasz.balcerzak@python-center.pl' == self.repo.get_changeset('3803844fdbd3').author_email
         assert '' == self.repo.get_changeset('84478366594b').author_email
     def test_author_username(self):
         assert 'Marcin Kuzminski' == self.repo.get_changeset('b986218ba1c9').author_name
         assert 'Lukasz Balcerzak' == self.repo.get_changeset('3803844fdbd3').author_name
         assert 'marcink' == self.repo.get_changeset('84478366594b').author_name
     def test_successors(self):
         init_chset = self.repo.get_changeset(0)
         assert init_chset.successors == []
     def test_predecessors(self):
         init_chset = self.repo.get_changeset(0)
         assert len(init_chset.predecessors) == 0

kallithea/tests/vcs/test_inmemchangesets.py

➞

Show inline comments

 # encoding: utf-8
 """
 Tests so called "in memory changesets" commit API of vcs.
 """
 import datetime
 import pytest
 from kallithea.lib.vcs.exceptions import (
     EmptyRepositoryError, NodeAlreadyAddedError, NodeAlreadyChangedError, NodeAlreadyExistsError, NodeAlreadyRemovedError, NodeDoesNotExistError, NodeNotChangedError)
 from kallithea.lib.vcs.nodes import DirNode, FileNode
 from kallithea.tests.vcs.base import _BackendTestMixin
 class InMemoryChangesetTestMixin(_BackendTestMixin):
     @classmethod
     def _get_commits(cls):
         # Note: this is slightly different than the regular _get_commits methods
         # as we don't actually return any commits. The creation of commits is
         # handled in the tests themselves.
         cls.nodes = [
             FileNode('foobar', content='Foo & bar'),
             FileNode('foobar2', content='Foo & bar, doubled!'),
             FileNode('foo bar with spaces', content=''),
             FileNode('foo/bar/baz', content='Inside'),
             FileNode('foo/bar/file.bin', content='\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00;\x00\x03\x00\xfe\xff\t\x00\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x1a\x00\x00\x00\x00\x00\x00\x00\x00\x10\x00\x00\x18\x00\x00\x00\x01\x00\x00\x00\xfe\xff\xff\xff\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff'),
+        ]
         commits = []
         return commits
     def test_add(self):
         rev_count = len(self.repo.revisions)
         to_add = [FileNode(node.path, content=node.content)
             for node in self.nodes]
         for node in to_add:
             self.imc.add(node)
         message = u'Added: %s' % ', '.join((node.path for node in self.nodes))
-        author = unicode(self.__class__)
+        author = str(self.__class__)
         changeset = self.imc.commit(message=message, author=author)
         newtip = self.repo.get_changeset()
         assert changeset == newtip
         assert rev_count + 1 == len(self.repo.revisions)
         assert newtip.message == message
         assert newtip.author == author
         assert not any((
             self.imc.added,
             self.imc.changed,
             self.imc.removed
         ))
         for node in to_add:
             assert newtip.get_node(node.path).content == node.content
     def test_add_in_bulk(self):
         rev_count = len(self.repo.revisions)
         to_add = [FileNode(node.path, content=node.content)
             for node in self.nodes]
         self.imc.add(*to_add)
         message = u'Added: %s' % ', '.join((node.path for node in self.nodes))
-        author = unicode(self.__class__)
+        author = str(self.__class__)
         changeset = self.imc.commit(message=message, author=author)
         newtip = self.repo.get_changeset()
         assert changeset == newtip
         assert rev_count + 1 == len(self.repo.revisions)
         assert newtip.message == message
         assert newtip.author == author
         assert not any((
             self.imc.added,
             self.imc.changed,
             self.imc.removed
         ))
         for node in to_add:
             assert newtip.get_node(node.path).content == node.content
     def test_add_actually_adds_all_nodes_at_second_commit_too(self):
         self.imc.add(FileNode('foo/bar/image.png', content='\0'))
         self.imc.add(FileNode('foo/README.txt', content='readme!'))
         changeset = self.imc.commit(u'Initial', u'joe.doe@example.com')
         assert isinstance(changeset.get_node('foo'), DirNode)
         assert isinstance(changeset.get_node('foo/bar'), DirNode)
         assert changeset.get_node('foo/bar/image.png').content == b'\0'
         assert changeset.get_node('foo/README.txt').content == b'readme!'
         # commit some more files again
         to_add = [
             FileNode('foo/bar/foobaz/bar', content='foo'),
             FileNode('foo/bar/another/bar', content='foo'),
             FileNode('foo/baz.txt', content='foo'),
             FileNode('foobar/foobaz/file', content='foo'),
             FileNode('foobar/barbaz', content='foo'),
+        ]
         self.imc.add(*to_add)
         changeset = self.imc.commit(u'Another', u'joe.doe@example.com')
         changeset.get_node('foo/bar/foobaz/bar').content == b'foo'
         changeset.get_node('foo/bar/another/bar').content == b'foo'
         changeset.get_node('foo/baz.txt').content == b'foo'
         changeset.get_node('foobar/foobaz/file').content == b'foo'
         changeset.get_node('foobar/barbaz').content == b'foo'
     def test_add_non_ascii_files(self):
         rev_count = len(self.repo.revisions)
         to_add = [
             FileNode('żółwik/zwierzątko', content='ćććć'),
             FileNode(u'żółwik/zwierzątko_uni', content=u'ćććć'),
+        ]
         for node in to_add:
             self.imc.add(node)
         message = u'Added: %s' % ', '.join((node.path for node in self.nodes))
-        author = unicode(self.__class__)
+        author = str(self.__class__)
         changeset = self.imc.commit(message=message, author=author)
         newtip = self.repo.get_changeset()
         assert changeset == newtip
         assert rev_count + 1 == len(self.repo.revisions)
         assert newtip.message == message
         assert newtip.author == author
         assert not any((
             self.imc.added,
             self.imc.changed,
             self.imc.removed
         ))
         for node in to_add:
             assert newtip.get_node(node.path).content == node.content
     def test_add_raise_already_added(self):
         node = FileNode('foobar', content='baz')
         self.imc.add(node)
         with pytest.raises(NodeAlreadyAddedError):
             self.imc.add(node)
     def test_check_integrity_raise_already_exist(self):
         node = FileNode('foobar', content='baz')
         self.imc.add(node)
-        self.imc.commit(message=u'Added foobar', author=unicode(self))
+        self.imc.commit(message=u'Added foobar', author=str(self))
         self.imc.add(node)
         with pytest.raises(NodeAlreadyExistsError):
             self.imc.commit(message='new message',
                             author=str(self))
     def test_change(self):
         self.imc.add(FileNode('foo/bar/baz', content='foo'))
         self.imc.add(FileNode('foo/fbar', content='foobar'))
         tip = self.imc.commit(u'Initial', u'joe.doe@example.com')
         # Change node's content
         node = FileNode('foo/bar/baz', content='My **changed** content')
         self.imc.change(node)
         self.imc.commit(u'Changed %s' % node.path, u'joe.doe@example.com')
         newtip = self.repo.get_changeset()
         assert tip != newtip
         assert tip.raw_id != newtip.raw_id
         assert newtip.get_node('foo/bar/baz').content == b'My **changed** content'
     def test_change_non_ascii(self):
         to_add = [
             FileNode('żółwik/zwierzątko', content='ćććć'),
             FileNode(u'żółwik/zwierzątko_uni', content=u'ćććć'),
+        ]
         for node in to_add:
             self.imc.add(node)
         tip = self.imc.commit(u'Initial', u'joe.doe@example.com')
         # Change node's content
         node = FileNode('żółwik/zwierzątko', content='My **changed** content')
         self.imc.change(node)
         self.imc.commit(u'Changed %s' % node.path, u'joe.doe@example.com')
         node = FileNode(u'żółwik/zwierzątko_uni', content=u'My **changed** content')
         self.imc.change(node)
         self.imc.commit(u'Changed %s' % node.path, u'joe.doe@example.com')
         newtip = self.repo.get_changeset()
         assert tip != newtip
         assert tip.raw_id != newtip.raw_id
         assert newtip.get_node('żółwik/zwierzątko').content == b'My **changed** content'
         assert newtip.get_node('żółwik/zwierzątko_uni').content == b'My **changed** content'
     def test_change_raise_empty_repository(self):
         node = FileNode('foobar')
         with pytest.raises(EmptyRepositoryError):
             self.imc.change(node)
     def test_check_integrity_change_raise_node_does_not_exist(self):
         node = FileNode('foobar', content='baz')
         self.imc.add(node)
-        self.imc.commit(message=u'Added foobar', author=unicode(self))
+        self.imc.commit(message=u'Added foobar', author=str(self))
         node = FileNode('not-foobar', content='')
         self.imc.change(node)
         with pytest.raises(NodeDoesNotExistError):
             self.imc.commit(message='Changed not existing node', author=str(self))
     def test_change_raise_node_already_changed(self):
         node = FileNode('foobar', content='baz')
         self.imc.add(node)
-        self.imc.commit(message=u'Added foobar', author=unicode(self))
+        self.imc.commit(message=u'Added foobar', author=str(self))
         node = FileNode('foobar', content='more baz')
         self.imc.change(node)
         with pytest.raises(NodeAlreadyChangedError):
             self.imc.change(node)
     def test_check_integrity_change_raise_node_not_changed(self):
         self.test_add()  # Performs first commit
         node = FileNode(self.nodes[0].path, content=self.nodes[0].content)
         self.imc.change(node)
         with pytest.raises(NodeNotChangedError):
             self.imc.commit(
                 message=u'Trying to mark node as changed without touching it',
-                author=unicode(self)
+                author=str(self),
+            )
     def test_change_raise_node_already_removed(self):
         node = FileNode('foobar', content='baz')
         self.imc.add(node)
-        self.imc.commit(message=u'Added foobar', author=unicode(self))
+        self.imc.commit(message=u'Added foobar', author=str(self))
         self.imc.remove(FileNode('foobar'))
         with pytest.raises(NodeAlreadyRemovedError):
             self.imc.change(node)
     def test_remove(self):
         self.test_add()  # Performs first commit
         tip = self.repo.get_changeset()
         node = self.nodes[0]
         assert node.content == tip.get_node(node.path).content
         self.imc.remove(node)
-        self.imc.commit(message=u'Removed %s' % node.path, author=unicode(self))
+        self.imc.commit(message=u'Removed %s' % node.path, author=str(self))
         newtip = self.repo.get_changeset()
         assert tip != newtip
         assert tip.raw_id != newtip.raw_id
         with pytest.raises(NodeDoesNotExistError):
             newtip.get_node(node.path)
     def test_remove_last_file_from_directory(self):
         node = FileNode('omg/qwe/foo/bar', content='foobar')
         self.imc.add(node)
         self.imc.commit(u'added', u'joe doe')
         self.imc.remove(node)
         tip = self.imc.commit(u'removed', u'joe doe')
         with pytest.raises(NodeDoesNotExistError):
             tip.get_node('omg/qwe/foo/bar')
     def test_remove_raise_node_does_not_exist(self):
         self.imc.remove(self.nodes[0])
         with pytest.raises(NodeDoesNotExistError):
             self.imc.commit(
                 message='Trying to remove node at empty repository',
                 author=str(self)
+                author=str(self),
+            )
     def test_check_integrity_remove_raise_node_does_not_exist(self):
         self.test_add()  # Performs first commit
         node = FileNode('no-such-file')
         self.imc.remove(node)
         with pytest.raises(NodeDoesNotExistError):
             self.imc.commit(
                 message=u'Trying to remove not existing node',
-                author=unicode(self)
+                author=str(self),
+            )
     def test_remove_raise_node_already_removed(self):
         self.test_add() # Performs first commit
         node = FileNode(self.nodes[0].path)
         self.imc.remove(node)
         with pytest.raises(NodeAlreadyRemovedError):
             self.imc.remove(node)
     def test_remove_raise_node_already_changed(self):
         self.test_add()  # Performs first commit
         node = FileNode(self.nodes[0].path, content='Bending time')
         self.imc.change(node)
         with pytest.raises(NodeAlreadyChangedError):
             self.imc.remove(node)
     def test_reset(self):
         self.imc.add(FileNode('foo', content='bar'))
         #self.imc.change(FileNode('baz', content='new'))
         #self.imc.remove(FileNode('qwe'))
         self.imc.reset()
         assert not any((
             self.imc.added,
             self.imc.changed,
             self.imc.removed
         ))
     def test_multiple_commits(self):
         N = 3  # number of commits to perform
         last = None
         for x in range(N):
             fname = 'file%s' % str(x).rjust(5, '0')
             content = 'foobar\n' * x
             node = FileNode(fname, content=content)
             self.imc.add(node)
             commit = self.imc.commit(u"Commit no. %s" % (x + 1), author=u'vcs')
             assert last != commit
             last = commit
         # Check commit number for same repo
         assert len(self.repo.revisions) == N
         # Check commit number for recreated repo
         assert len(self.repo.revisions) == N
     def test_date_attr(self):
         node = FileNode('foobar.txt', content='Foobared!')
         self.imc.add(node)
         date = datetime.datetime(1985, 1, 30, 1, 45)
         commit = self.imc.commit(u"Committed at time when I was born ;-)",
             author=u'lb <lb@example.com>', date=date)
         assert commit.date == date
 class TestGitInMemoryChangeset(InMemoryChangesetTestMixin):
     backend_alias = 'git'
 class TestHgInMemoryChangeset(InMemoryChangesetTestMixin):
     backend_alias = 'hg'

0 comments (0 inline, 0 general)