python3: fixed usage of stringIO module

rhodecode/apps/file_store/utils.py

@@ -1,58 +1,58 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
 import uuid
-import StringIO
+from io import StringIO
 import pathlib2
 
 
 def get_file_storage(settings):
     from rhodecode.apps.file_store.backends.local_store import LocalFileStorage
     from rhodecode.apps.file_store import config_keys
     store_path = settings.get(config_keys.store_path)
     return LocalFileStorage(base_path=store_path)
 
 
 def splitext(filename):
     ext = ''.join(pathlib2.Path(filename).suffixes)
     return filename, ext
 
 
 def uid_filename(filename, randomized=True):
     """
     Generates a randomized or stable (uuid) filename,
     preserving the original extension.
 
     :param filename: the original filename
     :param randomized: define if filename should be stable (sha1 based) or randomized
     """
 
     _, ext = splitext(filename)
     if randomized:
         uid = uuid.uuid4()
     else:
         hash_key = '{}.{}'.format(filename, 'store')
         uid = uuid.uuid5(uuid.NAMESPACE_URL, hash_key)
     return str(uid) + ext.lower()
 
 
 def bytes_to_file_obj(bytes_data):
     return StringIO.StringIO(bytes_data)

rhodecode/lib/_vendor/authomatic/six.py

@@ -1,839 +1,839 b''
 # -*- coding: utf-8 -*-
 """Utilities for writing code that runs on Python 2 and 3"""
 
 # Copyright (c) 2010-2015 Benjamin Peterson
 #
 # Permission is hereby granted, free of charge, to any person obtaining a copy
 # of this software and associated documentation files (the "Software"), to deal
 # in the Software without restriction, including without limitation the rights
 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 # copies of the Software, and to permit persons to whom the Software is
 # furnished to do so, subject to the following conditions:
 #
 # The above copyright notice and this permission notice shall be included in all
 # copies or substantial portions of the Software.
 #
 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
 
 
 
 import functools
 import itertools
 import operator
 import sys
 import types
 
 __author__ = "Benjamin Peterson <[e-mail]>"
 __version__ = "1.9.0"
 
 
 # Useful for very coarse version differentiation.
 PY2 = sys.version_info[0] == 2
 PY3 = sys.version_info[0] == 3
 
 if PY3:
     string_types = str,
     integer_types = int,
     class_types = type,
     text_type = str
     binary_type = bytes
 
     MAXSIZE = sys.maxsize
 else:
     string_types = str,
     integer_types = (int, long)
     class_types = (type, types.ClassType)
     text_type = unicode
     binary_type = str
 
     if sys.platform.startswith("java"):
         # Jython always uses 32 bits.
         MAXSIZE = int((1 << 31) - 1)
     else:
         # It's possible to have sizeof(long) != sizeof(Py_ssize_t).
         class X(object):
             def __len__(self):
                 return 1 << 31
         try:
             len(X())
         except OverflowError:
             # 32-bit
             MAXSIZE = int((1 << 31) - 1)
         else:
             # 64-bit
             MAXSIZE = int((1 << 63) - 1)
         del X
 
 
 def _add_doc(func, doc):
     """Add documentation to a function."""
     func.__doc__ = doc
 
 
 def _import_module(name):
     """Import module, returning the module after the last dot."""
     __import__(name)
     return sys.modules[name]
 
 
 class _LazyDescr(object):
 
     def __init__(self, name):
         self.name = name
 
     def __get__(self, obj, tp):
         result = self._resolve()
         setattr(obj, self.name, result) # Invokes __set__.
         try:
             # This is a bit ugly, but it avoids running this again by
             # removing this descriptor.
             delattr(obj.__class__, self.name)
         except AttributeError:
             pass
         return result
 
 
 class MovedModule(_LazyDescr):
 
     def __init__(self, name, old, new=None):
         super(MovedModule, self).__init__(name)
         if PY3:
             if new is None:
                 new = name
             self.mod = new
         else:
             self.mod = old
 
     def _resolve(self):
         return _import_module(self.mod)
 
     def __getattr__(self, attr):
         _module = self._resolve()
         value = getattr(_module, attr)
         setattr(self, attr, value)
         return value
 
 
 class _LazyModule(types.ModuleType):
 
     def __init__(self, name):
         super(_LazyModule, self).__init__(name)
         self.__doc__ = self.__class__.__doc__
 
     def __dir__(self):
         attrs = ["__doc__", "__name__"]
         attrs += [attr.name for attr in self._moved_attributes]
         return attrs
 
     # Subclasses should override this
     _moved_attributes = []
 
 
 class MovedAttribute(_LazyDescr):
 
     def __init__(self, name, old_mod, new_mod, old_attr=None, new_attr=None):
         super(MovedAttribute, self).__init__(name)
         if PY3:
             if new_mod is None:
                 new_mod = name
             self.mod = new_mod
             if new_attr is None:
                 if old_attr is None:
                     new_attr = name
                 else:
                     new_attr = old_attr
             self.attr = new_attr
         else:
             self.mod = old_mod
             if old_attr is None:
                 old_attr = name
             self.attr = old_attr
 
     def _resolve(self):
         module = _import_module(self.mod)
         return getattr(module, self.attr)
 
 
 class _SixMetaPathImporter(object):
     """
     A meta path importer to import six.moves and its submodules.
 
     This class implements a PEP302 finder and loader. It should be compatible
     with Python 2.5 and all existing versions of Python3
     """
     def __init__(self, six_module_name):
         self.name = six_module_name
         self.known_modules = {}
 
     def _add_module(self, mod, *fullnames):
         for fullname in fullnames:
             self.known_modules[self.name + "." + fullname] = mod
 
     def _get_module(self, fullname):
         return self.known_modules[self.name + "." + fullname]
 
     def find_module(self, fullname, path=None):
         if fullname in self.known_modules:
             return self
         return None
 
     def __get_module(self, fullname):
         try:
             return self.known_modules[fullname]
         except KeyError:
             raise ImportError("This loader does not know module " + fullname)
 
     def load_module(self, fullname):
         try:
             # in case of a reload
             return sys.modules[fullname]
         except KeyError:
             pass
         mod = self.__get_module(fullname)
         if isinstance(mod, MovedModule):
             mod = mod._resolve()
         else:
             mod.__loader__ = self
         sys.modules[fullname] = mod
         return mod
 
     def is_package(self, fullname):
         """
         Return true, if the named module is a package.
 
         We need this method to get correct spec objects with
         Python 3.4 (see PEP451)
         """
         return hasattr(self.__get_module(fullname), "__path__")
 
     def get_code(self, fullname):
         """Return None
 
         Required, if is_package is implemented"""
         self.__get_module(fullname)  # eventually raises ImportError
         return None
     get_source = get_code  # same as get_code
 
 _importer = _SixMetaPathImporter(__name__)
 
 
 class _MovedItems(_LazyModule):
     """Lazy loading of moved objects"""
     __path__ = []  # mark as package
 
 
 _moved_attributes = [
     MovedAttribute("cStringIO", "cStringIO", "io", "StringIO"),
     MovedAttribute("filter", "itertools", "builtins", "ifilter", "filter"),
     MovedAttribute("filterfalse", "itertools", "itertools", "ifilterfalse", "filterfalse"),
     MovedAttribute("input", "__builtin__", "builtins", "raw_input", "input"),
     MovedAttribute("intern", "__builtin__", "sys"),
     MovedAttribute("map", "itertools", "builtins", "imap", "map"),
     MovedAttribute("range", "__builtin__", "builtins", "xrange", "range"),
     MovedAttribute("reload_module", "__builtin__", "imp", "reload"),
     MovedAttribute("reduce", "__builtin__", "functools"),
     MovedAttribute("shlex_quote", "pipes", "shlex", "quote"),
     MovedAttribute("StringIO", "StringIO", "io"),
     MovedAttribute("UserDict", "UserDict", "collections"),
     MovedAttribute("UserList", "UserList", "collections"),
     MovedAttribute("UserString", "UserString", "collections"),
     MovedAttribute("xrange", "__builtin__", "builtins", "xrange", "range"),
     MovedAttribute("zip", "itertools", "builtins", "izip", "zip"),
     MovedAttribute("zip_longest", "itertools", "itertools", "izip_longest", "zip_longest"),
 
     MovedModule("builtins", "__builtin__"),
     MovedModule("configparser", "ConfigParser"),
     MovedModule("copyreg", "copy_reg"),
     MovedModule("dbm_gnu", "gdbm", "dbm.gnu"),
     MovedModule("_dummy_thread", "dummy_thread", "_dummy_thread"),
     MovedModule("http_cookiejar", "cookielib", "http.cookiejar"),
     MovedModule("http_cookies", "Cookie", "http.cookies"),
     MovedModule("html_entities", "htmlentitydefs", "html.entities"),
     MovedModule("html_parser", "HTMLParser", "html.parser"),
     MovedModule("http_client", "httplib", "http.client"),
     MovedModule("email_mime_multipart", "email.MIMEMultipart", "email.mime.multipart"),
     MovedModule("email_mime_nonmultipart", "email.MIMENonMultipart", "email.mime.nonmultipart"),
     MovedModule("email_mime_text", "email.MIMEText", "email.mime.text"),
     MovedModule("email_mime_base", "email.MIMEBase", "email.mime.base"),
     MovedModule("BaseHTTPServer", "BaseHTTPServer", "http.server"),
     MovedModule("CGIHTTPServer", "CGIHTTPServer", "http.server"),
     MovedModule("SimpleHTTPServer", "SimpleHTTPServer", "http.server"),
     MovedModule("cPickle", "cPickle", "pickle"),
     MovedModule("queue", "Queue"),
     MovedModule("reprlib", "repr"),
     MovedModule("socketserver", "SocketServer"),
     MovedModule("_thread", "thread", "_thread"),
     MovedModule("tkinter", "Tkinter"),
     MovedModule("tkinter_dialog", "Dialog", "tkinter.dialog"),
     MovedModule("tkinter_filedialog", "FileDialog", "tkinter.filedialog"),
     MovedModule("tkinter_scrolledtext", "ScrolledText", "tkinter.scrolledtext"),
     MovedModule("tkinter_simpledialog", "SimpleDialog", "tkinter.simpledialog"),
     MovedModule("tkinter_tix", "Tix", "tkinter.tix"),
     MovedModule("tkinter_ttk", "ttk", "tkinter.ttk"),
     MovedModule("tkinter_constants", "Tkconstants", "tkinter.constants"),
     MovedModule("tkinter_dnd", "Tkdnd", "tkinter.dnd"),
     MovedModule("tkinter_colorchooser", "tkColorChooser",
                 "tkinter.colorchooser"),
     MovedModule("tkinter_commondialog", "tkCommonDialog",
                 "tkinter.commondialog"),
     MovedModule("tkinter_tkfiledialog", "tkFileDialog", "tkinter.filedialog"),
     MovedModule("tkinter_font", "tkFont", "tkinter.font"),
     MovedModule("tkinter_messagebox", "tkMessageBox", "tkinter.messagebox"),
     MovedModule("tkinter_tksimpledialog", "tkSimpleDialog",
                 "tkinter.simpledialog"),
     MovedModule("urllib_parse", __name__ + ".moves.urllib_parse", "urllib.parse"),
     MovedModule("urllib_error", __name__ + ".moves.urllib_error", "urllib.error"),
     MovedModule("urllib", __name__ + ".moves.urllib", __name__ + ".moves.urllib"),
     MovedModule("urllib_robotparser", "robotparser", "urllib.robotparser"),
     MovedModule("xmlrpc_client", "xmlrpclib", "xmlrpc.client"),
     MovedModule("xmlrpc_server", "SimpleXMLRPCServer", "xmlrpc.server"),
     MovedModule("winreg", "_winreg"),
 ]
 for attr in _moved_attributes:
     setattr(_MovedItems, attr.name, attr)
     if isinstance(attr, MovedModule):
         _importer._add_module(attr, "moves." + attr.name)
 del attr
 
 _MovedItems._moved_attributes = _moved_attributes
 
 moves = _MovedItems(__name__ + ".moves")
 _importer._add_module(moves, "moves")
 
 
 class Module_six_moves_urllib_parse(_LazyModule):
     """Lazy loading of moved objects in six.moves.urllib_parse"""
 
 
 _urllib_parse_moved_attributes = [
     MovedAttribute("ParseResult", "urlparse", "urllib.parse"),
     MovedAttribute("SplitResult", "urlparse", "urllib.parse"),
     MovedAttribute("parse_qs", "urlparse", "urllib.parse"),
     MovedAttribute("parse_qsl", "urlparse", "urllib.parse"),
     MovedAttribute("urldefrag", "urlparse", "urllib.parse"),
     MovedAttribute("urljoin", "urlparse", "urllib.parse"),
     MovedAttribute("urlparse", "urlparse", "urllib.parse"),
     MovedAttribute("urlsplit", "urlparse", "urllib.parse"),
     MovedAttribute("urlunparse", "urlparse", "urllib.parse"),
     MovedAttribute("urlunsplit", "urlparse", "urllib.parse"),
     MovedAttribute("quote", "urllib", "urllib.parse"),
     MovedAttribute("quote_plus", "urllib", "urllib.parse"),
     MovedAttribute("unquote", "urllib", "urllib.parse"),
     MovedAttribute("unquote_plus", "urllib", "urllib.parse"),
     MovedAttribute("urlencode", "urllib", "urllib.parse"),
     MovedAttribute("splitquery", "urllib", "urllib.parse"),
     MovedAttribute("splittag", "urllib", "urllib.parse"),
     MovedAttribute("splituser", "urllib", "urllib.parse"),
     MovedAttribute("uses_fragment", "urlparse", "urllib.parse"),
     MovedAttribute("uses_netloc", "urlparse", "urllib.parse"),
     MovedAttribute("uses_params", "urlparse", "urllib.parse"),
     MovedAttribute("uses_query", "urlparse", "urllib.parse"),
     MovedAttribute("uses_relative", "urlparse", "urllib.parse"),
 ]
 for attr in _urllib_parse_moved_attributes:
     setattr(Module_six_moves_urllib_parse, attr.name, attr)
 del attr
 
 Module_six_moves_urllib_parse._moved_attributes = _urllib_parse_moved_attributes
 
 _importer._add_module(Module_six_moves_urllib_parse(__name__ + ".moves.urllib_parse"),
                       "moves.urllib_parse", "moves.urllib.parse")
 
 
 class Module_six_moves_urllib_error(_LazyModule):
     """Lazy loading of moved objects in six.moves.urllib_error"""
 
 
 _urllib_error_moved_attributes = [
     MovedAttribute("URLError", "urllib2", "urllib.error"),
     MovedAttribute("HTTPError", "urllib2", "urllib.error"),
     MovedAttribute("ContentTooShortError", "urllib", "urllib.error"),
 ]
 for attr in _urllib_error_moved_attributes:
     setattr(Module_six_moves_urllib_error, attr.name, attr)
 del attr
 
 Module_six_moves_urllib_error._moved_attributes = _urllib_error_moved_attributes
 
 _importer._add_module(Module_six_moves_urllib_error(__name__ + ".moves.urllib.error"),
                       "moves.urllib_error", "moves.urllib.error")
 
 
 class Module_six_moves_urllib_request(_LazyModule):
     """Lazy loading of moved objects in six.moves.urllib_request"""
 
 
 _urllib_request_moved_attributes = [
     MovedAttribute("urlopen", "urllib2", "urllib.request"),
     MovedAttribute("install_opener", "urllib2", "urllib.request"),
     MovedAttribute("build_opener", "urllib2", "urllib.request"),
     MovedAttribute("pathname2url", "urllib", "urllib.request"),
     MovedAttribute("url2pathname", "urllib", "urllib.request"),
     MovedAttribute("getproxies", "urllib", "urllib.request"),
     MovedAttribute("Request", "urllib2", "urllib.request"),
     MovedAttribute("OpenerDirector", "urllib2", "urllib.request"),
     MovedAttribute("HTTPDefaultErrorHandler", "urllib2", "urllib.request"),
     MovedAttribute("HTTPRedirectHandler", "urllib2", "urllib.request"),
     MovedAttribute("HTTPCookieProcessor", "urllib2", "urllib.request"),
     MovedAttribute("ProxyHandler", "urllib2", "urllib.request"),
     MovedAttribute("BaseHandler", "urllib2", "urllib.request"),
     MovedAttribute("HTTPPasswordMgr", "urllib2", "urllib.request"),
     MovedAttribute("HTTPPasswordMgrWithDefaultRealm", "urllib2", "urllib.request"),
     MovedAttribute("AbstractBasicAuthHandler", "urllib2", "urllib.request"),
     MovedAttribute("HTTPBasicAuthHandler", "urllib2", "urllib.request"),
     MovedAttribute("ProxyBasicAuthHandler", "urllib2", "urllib.request"),
     MovedAttribute("AbstractDigestAuthHandler", "urllib2", "urllib.request"),
     MovedAttribute("HTTPDigestAuthHandler", "urllib2", "urllib.request"),
     MovedAttribute("ProxyDigestAuthHandler", "urllib2", "urllib.request"),
     MovedAttribute("HTTPHandler", "urllib2", "urllib.request"),
     MovedAttribute("HTTPSHandler", "urllib2", "urllib.request"),
     MovedAttribute("FileHandler", "urllib2", "urllib.request"),
     MovedAttribute("FTPHandler", "urllib2", "urllib.request"),
     MovedAttribute("CacheFTPHandler", "urllib2", "urllib.request"),
     MovedAttribute("UnknownHandler", "urllib2", "urllib.request"),
     MovedAttribute("HTTPErrorProcessor", "urllib2", "urllib.request"),
     MovedAttribute("urlretrieve", "urllib", "urllib.request"),
     MovedAttribute("urlcleanup", "urllib", "urllib.request"),
     MovedAttribute("URLopener", "urllib", "urllib.request"),
     MovedAttribute("FancyURLopener", "urllib", "urllib.request"),
     MovedAttribute("proxy_bypass", "urllib", "urllib.request"),
 ]
 for attr in _urllib_request_moved_attributes:
     setattr(Module_six_moves_urllib_request, attr.name, attr)
 del attr
 
 Module_six_moves_urllib_request._moved_attributes = _urllib_request_moved_attributes
 
 _importer._add_module(Module_six_moves_urllib_request(__name__ + ".moves.urllib.request"),
                       "moves.urllib_request", "moves.urllib.request")
 
 
 class Module_six_moves_urllib_response(_LazyModule):
     """Lazy loading of moved objects in six.moves.urllib_response"""
 
 
 _urllib_response_moved_attributes = [
     MovedAttribute("addbase", "urllib", "urllib.response"),
     MovedAttribute("addclosehook", "urllib", "urllib.response"),
     MovedAttribute("addinfo", "urllib", "urllib.response"),
     MovedAttribute("addinfourl", "urllib", "urllib.response"),
 ]
 for attr in _urllib_response_moved_attributes:
     setattr(Module_six_moves_urllib_response, attr.name, attr)
 del attr
 
 Module_six_moves_urllib_response._moved_attributes = _urllib_response_moved_attributes
 
 _importer._add_module(Module_six_moves_urllib_response(__name__ + ".moves.urllib.response"),
                       "moves.urllib_response", "moves.urllib.response")
 
 
 class Module_six_moves_urllib_robotparser(_LazyModule):
     """Lazy loading of moved objects in six.moves.urllib_robotparser"""
 
 
 _urllib_robotparser_moved_attributes = [
     MovedAttribute("RobotFileParser", "robotparser", "urllib.robotparser"),
 ]
 for attr in _urllib_robotparser_moved_attributes:
     setattr(Module_six_moves_urllib_robotparser, attr.name, attr)
 del attr
 
 Module_six_moves_urllib_robotparser._moved_attributes = _urllib_robotparser_moved_attributes
 
 _importer._add_module(Module_six_moves_urllib_robotparser(__name__ + ".moves.urllib.robotparser"),
                       "moves.urllib_robotparser", "moves.urllib.robotparser")
 
 
 class Module_six_moves_urllib(types.ModuleType):
     """Create a six.moves.urllib namespace that resembles the Python 3 namespace"""
     __path__ = []  # mark as package
     parse = _importer._get_module("moves.urllib_parse")
     error = _importer._get_module("moves.urllib_error")
     request = _importer._get_module("moves.urllib_request")
     response = _importer._get_module("moves.urllib_response")
     robotparser = _importer._get_module("moves.urllib_robotparser")
 
     def __dir__(self):
         return ['parse', 'error', 'request', 'response', 'robotparser']
 
 _importer._add_module(Module_six_moves_urllib(__name__ + ".moves.urllib"),
                       "moves.urllib")
 
 
 def add_move(move):
     """Add an item to six.moves."""
     setattr(_MovedItems, move.name, move)
 
 
 def remove_move(name):
     """Remove item from six.moves."""
     try:
         delattr(_MovedItems, name)
     except AttributeError:
         try:
             del moves.__dict__[name]
         except KeyError:
             raise AttributeError("no such move, %r" % (name,))
 
 
 if PY3:
     _meth_func = "__func__"
     _meth_self = "__self__"
 
     _func_closure = "__closure__"
     _func_code = "__code__"
     _func_defaults = "__defaults__"
     _func_globals = "__globals__"
 else:
     _meth_func = "im_func"
     _meth_self = "im_self"
 
     _func_closure = "func_closure"
     _func_code = "func_code"
     _func_defaults = "func_defaults"
     _func_globals = "func_globals"
 
 
 try:
     advance_iterator = next
 except NameError:
     def advance_iterator(it):
         return it.next()
 next = advance_iterator
 
 
 try:
     callable = callable
 except NameError:
     def callable(obj):
         return any("__call__" in klass.__dict__ for klass in type(obj).__mro__)
 
 
 if PY3:
     def get_unbound_function(unbound):
         return unbound
 
     create_bound_method = types.MethodType
 
     Iterator = object
 else:
     def get_unbound_function(unbound):
         return unbound.im_func
 
     def create_bound_method(func, obj):
         return types.MethodType(func, obj, obj.__class__)
 
     class Iterator(object):
 
         def next(self):
             return type(self).__next__(self)
 
     callable = callable
 _add_doc(get_unbound_function,
          """Get the function out of a possibly unbound function""")
 
 
 get_method_function = operator.attrgetter(_meth_func)
 get_method_self = operator.attrgetter(_meth_self)
 get_function_closure = operator.attrgetter(_func_closure)
 get_function_code = operator.attrgetter(_func_code)
 get_function_defaults = operator.attrgetter(_func_defaults)
 get_function_globals = operator.attrgetter(_func_globals)
 
 
 if PY3:
     def iterkeys(d, **kw):
         return iter(d.keys(**kw))
 
     def itervalues(d, **kw):
         return iter(d.values(**kw))
 
     def iteritems(d, **kw):
         return iter(d.items(**kw))
 
     def iterlists(d, **kw):
         return iter(d.lists(**kw))
 
     viewkeys = operator.methodcaller("keys")
 
     viewvalues = operator.methodcaller("values")
 
     viewitems = operator.methodcaller("items")
 else:
     def iterkeys(d, **kw):
         return iter(d.iterkeys(**kw))
 
     def itervalues(d, **kw):
         return iter(d.itervalues(**kw))
 
     def iteritems(d, **kw):
         return iter(d.iteritems(**kw))
 
     def iterlists(d, **kw):
         return iter(d.iterlists(**kw))
 
     viewkeys = operator.methodcaller("viewkeys")
 
     viewvalues = operator.methodcaller("viewvalues")
 
     viewitems = operator.methodcaller("viewitems")
 
 _add_doc(iterkeys, "Return an iterator over the keys of a dictionary.")
 _add_doc(itervalues, "Return an iterator over the values of a dictionary.")
 _add_doc(iteritems,
          "Return an iterator over the (key, value) pairs of a dictionary.")
 _add_doc(iterlists,
          "Return an iterator over the (key, [values]) pairs of a dictionary.")
 
 
 if PY3:
     def b(s):
         return s.encode("latin-1")
     def u(s):
         return s
     unichr = chr
     if sys.version_info[1] <= 1:
         def int2byte(i):
             return bytes((i,))
     else:
         # This is about 2x faster than the implementation above on 3.2+
         int2byte = operator.methodcaller("to_bytes", 1, "big")
     byte2int = operator.itemgetter(0)
     indexbytes = operator.getitem
     iterbytes = iter
     import io
     StringIO = io.StringIO
     BytesIO = io.BytesIO
     _assertCountEqual = "assertCountEqual"
     _assertRaisesRegex = "assertRaisesRegex"
     _assertRegex = "assertRegex"
 else:
     def b(s):
         return s
     # Workaround for standalone backslash
     def u(s):
         return unicode(s.replace(r'\\', r'\\\\'), "unicode_escape")
     unichr = unichr
     int2byte = chr
     def byte2int(bs):
         return ord(bs[0])
     def indexbytes(buf, i):
         return ord(buf[i])
     iterbytes = functools.partial(itertools.imap, ord)
-    import StringIO
+    from io import StringIO
     StringIO = BytesIO = StringIO.StringIO
     _assertCountEqual = "assertItemsEqual"
     _assertRaisesRegex = "assertRaisesRegexp"
     _assertRegex = "assertRegexpMatches"
 _add_doc(b, """Byte literal""")
 _add_doc(u, """Text literal""")
 
 
 def assertCountEqual(self, *args, **kwargs):
     return getattr(self, _assertCountEqual)(*args, **kwargs)
 
 
 def assertRaisesRegex(self, *args, **kwargs):
     return getattr(self, _assertRaisesRegex)(*args, **kwargs)
 
 
 def assertRegex(self, *args, **kwargs):
     return getattr(self, _assertRegex)(*args, **kwargs)
 
 
 if PY3:
     exec_ = getattr(moves.builtins, "exec")
 
 
     def reraise(tp, value, tb=None):
         if value is None:
             value = tp()
         if value.__traceback__ is not tb:
             raise value.with_traceback(tb)
         raise value
 
 else:
     def exec_(_code_, _globs_=None, _locs_=None):
         """Execute code in a namespace."""
         if _globs_ is None:
             frame = sys._getframe(1)
             _globs_ = frame.f_globals
             if _locs_ is None:
                 _locs_ = frame.f_locals
             del frame
         elif _locs_ is None:
             _locs_ = _globs_
         exec("""exec _code_ in _globs_, _locs_""")
 
 
     exec_("""def reraise(tp, value, tb=None):
     raise tp, value, tb
 """)
 
 
 if sys.version_info[:2] == (3, 2):
     exec_("""def raise_from(value, from_value):
     if from_value is None:
         raise value
     raise value from from_value
 """)
 elif sys.version_info[:2] > (3, 2):
     exec_("""def raise_from(value, from_value):
     raise value from from_value
 """)
 else:
     def raise_from(value, from_value):
         raise value
 
 
 print_ = getattr(moves.builtins, "print", None)
 if print_ is None:
     def print_(*args, **kwargs):
         """The new-style print function for Python 2.4 and 2.5."""
         fp = kwargs.pop("file", sys.stdout)
         if fp is None:
             return
         def write(data):
             if not isinstance(data, str):
                 data = str(data)
             # If the file has an encoding, encode unicode with it.
             if (isinstance(fp, file) and
                     isinstance(data, unicode) and
                     fp.encoding is not None):
                 errors = getattr(fp, "errors", None)
                 if errors is None:
                     errors = "strict"
                 data = data.encode(fp.encoding, errors)
             fp.write(data)
         want_unicode = False
         sep = kwargs.pop("sep", None)
         if sep is not None:
             if isinstance(sep, unicode):
                 want_unicode = True
             elif not isinstance(sep, str):
                 raise TypeError("sep must be None or a string")
         end = kwargs.pop("end", None)
         if end is not None:
             if isinstance(end, unicode):
                 want_unicode = True
             elif not isinstance(end, str):
                 raise TypeError("end must be None or a string")
         if kwargs:
             raise TypeError("invalid keyword arguments to print()")
         if not want_unicode:
             for arg in args:
                 if isinstance(arg, unicode):
                     want_unicode = True
                     break
         if want_unicode:
             newline = unicode("\n")
             space = unicode(" ")
         else:
             newline = "\n"
             space = " "
         if sep is None:
             sep = space
         if end is None:
             end = newline
         for i, arg in enumerate(args):
             if i:
                 write(sep)
             write(arg)
         write(end)
 if sys.version_info[:2] < (3, 3):
     _print = print_
     def print_(*args, **kwargs):
         fp = kwargs.get("file", sys.stdout)
         flush = kwargs.pop("flush", False)
         _print(*args, **kwargs)
         if flush and fp is not None:
             fp.flush()
 
 _add_doc(reraise, """Reraise an exception.""")
 
 if sys.version_info[0:2] < (3, 4):
     def wraps(wrapped, assigned=functools.WRAPPER_ASSIGNMENTS,
               updated=functools.WRAPPER_UPDATES):
         def wrapper(f):
             f = functools.wraps(wrapped, assigned, updated)(f)
             f.__wrapped__ = wrapped
             return f
         return wrapper
 else:
     wraps = functools.wraps
 
 def with_metaclass(meta, *bases):
     """Create a base class with a metaclass."""
     # This requires a bit of explanation: the basic idea is to make a dummy
     # metaclass for one level of class instantiation that replaces itself with
     # the actual metaclass.
     class metaclass(meta):
         def __new__(cls, name, this_bases, d):
             return meta(name, bases, d)
     return type.__new__(metaclass, 'temporary_class', (), {})
 
 
 def add_metaclass(metaclass):
     """Class decorator for creating a class with a metaclass."""
     def wrapper(cls):
         orig_vars = cls.__dict__.copy()
         slots = orig_vars.get('__slots__')
         if slots is not None:
             if isinstance(slots, str):
                 slots = [slots]
             for slots_var in slots:
                 orig_vars.pop(slots_var)
         orig_vars.pop('__dict__', None)
         orig_vars.pop('__weakref__', None)
         return metaclass(cls.__name__, cls.__bases__, orig_vars)
     return wrapper
 
 
 def python_2_unicode_compatible(klass):
     """
     A decorator that defines __unicode__ and __str__ methods under Python 2.
     Under Python 3 it does nothing.
 
     To support Python 2 and 3 with a single code base, define a __str__ method
     returning text and apply this decorator to the class.
     """
     if PY2:
         if '__str__' not in klass.__dict__:
             raise ValueError("@python_2_unicode_compatible cannot be applied "
                              "to %s because it doesn't define __str__()." %
                              klass.__name__)
         klass.__unicode__ = klass.__str__
         klass.__str__ = lambda self: self.__unicode__().encode('utf-8')
     return klass
 
 
 # Complete the moves implementation.
 # This code is at the end of this module to speed up module loading.
 # Turn this module into a package.
 __path__ = []  # required for PEP 302 and PEP 451
 __package__ = __name__  # see PEP 366 @ReservedAssignment
 if globals().get("__spec__") is not None:
     __spec__.submodule_search_locations = []  # PEP 451 @UndefinedVariable
 # Remove other six meta path importers, since they cause problems. This can
 # happen if six is removed from sys.modules and then reloaded. (Setuptools does
 # this for some reason.)
 if sys.meta_path:
     for i, importer in enumerate(sys.meta_path):
         # Here's some real nastiness: Another "instance" of the six module might
         # be floating around. Therefore, we can't use isinstance() to check for
         # the six meta path importer, since the other six instance will have
         # inserted an importer with different class.
         if (type(importer).__name__ == "_SixMetaPathImporter" and
                 importer.name == __name__):
             del sys.meta_path[i]
             break
     del i, importer
 # Finally, add the importer to the meta path import hook.
 sys.meta_path.append(_importer)

rhodecode/lib/dbmigrate/migrate/changeset/ansisql.py

@@ -1,314 +1,314 b''
 """
    Extensions to SQLAlchemy for altering existing tables.
 
    At the moment, this isn't so much based off of ANSI as much as
    things that just happen to work with multiple databases.
 """
-import StringIO
+from io import StringIO
 
 import sqlalchemy as sa
 from sqlalchemy.schema import SchemaVisitor
 from sqlalchemy.engine.default import DefaultDialect
 from sqlalchemy.sql import ClauseElement
 from sqlalchemy.schema import (ForeignKeyConstraint,
                                PrimaryKeyConstraint,
                                CheckConstraint,
                                UniqueConstraint,
                                Index)
 
 import sqlalchemy.sql.compiler
 from rhodecode.lib.dbmigrate.migrate import exceptions
 from rhodecode.lib.dbmigrate.migrate.changeset import constraint
 from rhodecode.lib.dbmigrate.migrate.changeset import util
 
 from sqlalchemy.schema import AddConstraint, DropConstraint
 from sqlalchemy.sql.compiler import DDLCompiler
 SchemaGenerator = SchemaDropper = DDLCompiler
 
 
 class AlterTableVisitor(SchemaVisitor):
     """Common operations for ``ALTER TABLE`` statements."""
 
     # engine.Compiler looks for .statement
     # when it spawns off a new compiler
     statement = ClauseElement()
 
     def append(self, s):
         """Append content to the SchemaIterator's query buffer."""
 
         self.buffer.write(s)
 
     def execute(self):
         """Execute the contents of the SchemaIterator's buffer."""
         try:
             return self.connection.execute(self.buffer.getvalue())
         finally:
             self.buffer.seek(0)
             self.buffer.truncate()
 
     def __init__(self, dialect, connection, **kw):
         self.connection = connection
         self.buffer = StringIO.StringIO()
         self.preparer = dialect.identifier_preparer
         self.dialect = dialect
 
     def traverse_single(self, elem):
         ret = super(AlterTableVisitor, self).traverse_single(elem)
         if ret:
             # adapt to 0.6 which uses a string-returning
             # object
             self.append(" %s" % ret)
 
     def _to_table(self, param):
         """Returns the table object for the given param object."""
         if isinstance(param, (sa.Column, sa.Index, sa.schema.Constraint)):
             ret = param.table
         else:
             ret = param
         return ret
 
     def start_alter_table(self, param):
         """Returns the start of an ``ALTER TABLE`` SQL-Statement.
 
         Use the param object to determine the table name and use it
         for building the SQL statement.
 
         :param param: object to determine the table from
         :type param: :class:`sqlalchemy.Column`, :class:`sqlalchemy.Index`,
           :class:`sqlalchemy.schema.Constraint`, :class:`sqlalchemy.Table`,
           or string (table name)
         """
         table = self._to_table(param)
         self.append('\nALTER TABLE %s ' % self.preparer.format_table(table))
         return table
 
 
 class ANSIColumnGenerator(AlterTableVisitor, SchemaGenerator):
     """Extends ansisql generator for column creation (alter table add col)"""
 
     def visit_column(self, column):
         """Create a column (table already exists).
 
         :param column: column object
         :type column: :class:`sqlalchemy.Column` instance
         """
         if column.default is not None:
             self.traverse_single(column.default)
 
         table = self.start_alter_table(column)
         self.append("ADD ")
         self.append(self.get_column_specification(column))
 
         for cons in column.constraints:
             self.traverse_single(cons)
         self.execute()
 
         # ALTER TABLE STATEMENTS
 
         # add indexes and unique constraints
         if column.index_name:
             Index(column.index_name,column).create()
         elif column.unique_name:
             constraint.UniqueConstraint(column,
                                         name=column.unique_name).create()
 
         # SA bounds FK constraints to table, add manually
         for fk in column.foreign_keys:
             self.add_foreignkey(fk.constraint)
 
         # add primary key constraint if needed
         if column.primary_key_name:
             cons = constraint.PrimaryKeyConstraint(column,
                                                    name=column.primary_key_name)
             cons.create()
 
     def add_foreignkey(self, fk):
         self.connection.execute(AddConstraint(fk))
 
 class ANSIColumnDropper(AlterTableVisitor, SchemaDropper):
     """Extends ANSI SQL dropper for column dropping (``ALTER TABLE
     DROP COLUMN``).
     """
 
     def visit_column(self, column):
         """Drop a column from its table.
 
         :param column: the column object
         :type column: :class:`sqlalchemy.Column`
         """
         table = self.start_alter_table(column)
         self.append('DROP COLUMN %s' % self.preparer.format_column(column))
         self.execute()
 
 
 class ANSISchemaChanger(AlterTableVisitor, SchemaGenerator):
     """Manages changes to existing schema elements.
 
     Note that columns are schema elements; ``ALTER TABLE ADD COLUMN``
     is in SchemaGenerator.
 
     All items may be renamed. Columns can also have many of their properties -
     type, for example - changed.
 
     Each function is passed a tuple, containing (object, name); where
     object is a type of object you'd expect for that function
     (ie. table for visit_table) and name is the object's new
     name. NONE means the name is unchanged.
     """
 
     def visit_table(self, table):
         """Rename a table. Other ops aren't supported."""
         self.start_alter_table(table)
         q = util.safe_quote(table)
         self.append("RENAME TO %s" % self.preparer.quote(table.new_name, q))
         self.execute()
 
     def visit_index(self, index):
         """Rename an index"""
         if hasattr(self, '_validate_identifier'):
             # SA <= 0.6.3
             self.append("ALTER INDEX %s RENAME TO %s" % (
                     self.preparer.quote(
                         self._validate_identifier(
                             index.name, True), index.quote),
                     self.preparer.quote(
                         self._validate_identifier(
                             index.new_name, True), index.quote)))
         elif hasattr(self, '_index_identifier'):
             # SA >= 0.6.5, < 0.8
             self.append("ALTER INDEX %s RENAME TO %s" % (
                     self.preparer.quote(
                         self._index_identifier(
                             index.name), index.quote),
                     self.preparer.quote(
                         self._index_identifier(
                             index.new_name), index.quote)))
         else:
             # SA >= 0.8
             class NewName(object):
                 """Map obj.name -> obj.new_name"""
                 def __init__(self, index):
                     self.name = index.new_name
                     self._obj = index
 
                 def __getattr__(self, attr):
                     if attr == 'name':
                         return getattr(self, attr)
                     return getattr(self._obj, attr)
 
             self.append("ALTER INDEX %s RENAME TO %s" % (
                     self._prepared_index_name(index),
                     self._prepared_index_name(NewName(index))))
 
         self.execute()
 
     def visit_column(self, delta):
         """Rename/change a column."""
         # ALTER COLUMN is implemented as several ALTER statements
         keys = delta.keys()
         if 'type' in keys:
             self._run_subvisit(delta, self._visit_column_type)
         if 'nullable' in keys:
             self._run_subvisit(delta, self._visit_column_nullable)
         if 'server_default' in keys:
             # Skip 'default': only handle server-side defaults, others
             # are managed by the app, not the db.
             self._run_subvisit(delta, self._visit_column_default)
         if 'name' in keys:
             self._run_subvisit(delta, self._visit_column_name, start_alter=False)
 
     def _run_subvisit(self, delta, func, start_alter=True):
         """Runs visit method based on what needs to be changed on column"""
         table = self._to_table(delta.table)
         col_name = delta.current_name
         if start_alter:
             self.start_alter_column(table, col_name)
         ret = func(table, delta.result_column, delta)
         self.execute()
 
     def start_alter_column(self, table, col_name):
         """Starts ALTER COLUMN"""
         self.start_alter_table(table)
         q = util.safe_quote(table)
         self.append("ALTER COLUMN %s " % self.preparer.quote(col_name, q))
 
     def _visit_column_nullable(self, table, column, delta):
         nullable = delta['nullable']
         if nullable:
             self.append("DROP NOT NULL")
         else:
             self.append("SET NOT NULL")
 
     def _visit_column_default(self, table, column, delta):
         default_text = self.get_column_default_string(column)
         if default_text is not None:
             self.append("SET DEFAULT %s" % default_text)
         else:
             self.append("DROP DEFAULT")
 
     def _visit_column_type(self, table, column, delta):
         type_ = delta['type']
         type_text = str(type_.compile(dialect=self.dialect))
         self.append("TYPE %s" % type_text)
 
     def _visit_column_name(self, table, column, delta):
         self.start_alter_table(table)
         q = util.safe_quote(table)
         col_name = self.preparer.quote(delta.current_name, q)
         new_name = self.preparer.format_column(delta.result_column)
         self.append('RENAME COLUMN %s TO %s' % (col_name, new_name))
 
 
 class ANSIConstraintCommon(AlterTableVisitor):
     """
     Migrate's constraints require a separate creation function from
     SA's: Migrate's constraints are created independently of a table;
     SA's are created at the same time as the table.
     """
 
     def get_constraint_name(self, cons):
         """Gets a name for the given constraint.
 
         If the name is already set it will be used otherwise the
         constraint's :meth:`autoname <migrate.changeset.constraint.ConstraintChangeset.autoname>`
         method is used.
 
         :param cons: constraint object
         """
         if cons.name is not None:
             ret = cons.name
         else:
             ret = cons.name = cons.autoname()
         return self.preparer.quote(ret, cons.quote)
 
     def visit_migrate_primary_key_constraint(self, *p, **k):
         self._visit_constraint(*p, **k)
 
     def visit_migrate_foreign_key_constraint(self, *p, **k):
         self._visit_constraint(*p, **k)
 
     def visit_migrate_check_constraint(self, *p, **k):
         self._visit_constraint(*p, **k)
 
     def visit_migrate_unique_constraint(self, *p, **k):
         self._visit_constraint(*p, **k)
 
 class ANSIConstraintGenerator(ANSIConstraintCommon, SchemaGenerator):
     def _visit_constraint(self, constraint):
         constraint.name = self.get_constraint_name(constraint)
         self.append(self.process(AddConstraint(constraint)))
         self.execute()
 
 class ANSIConstraintDropper(ANSIConstraintCommon, SchemaDropper):
     def _visit_constraint(self, constraint):
         constraint.name = self.get_constraint_name(constraint)
         self.append(self.process(DropConstraint(constraint, cascade=constraint.cascade)))
         self.execute()
 
 
 class ANSIDialect(DefaultDialect):
     columngenerator = ANSIColumnGenerator
     columndropper = ANSIColumnDropper
     schemachanger = ANSISchemaChanger
     constraintgenerator = ANSIConstraintGenerator
     constraintdropper = ANSIConstraintDropper

rhodecode/lib/dbmigrate/migrate/versioning/script/py.py

@@ -1,159 +1,159 b''
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 
 import shutil
 import warnings
 import logging
 import inspect
-from StringIO import StringIO
+from io import StringIO
 
 from rhodecode.lib.dbmigrate import migrate
 from rhodecode.lib.dbmigrate.migrate.versioning import genmodel, schemadiff
 from rhodecode.lib.dbmigrate.migrate.versioning.config import operations
 from rhodecode.lib.dbmigrate.migrate.versioning.template import Template
 from rhodecode.lib.dbmigrate.migrate.versioning.script import base
 from rhodecode.lib.dbmigrate.migrate.versioning.util import import_path, load_model, with_engine
 from rhodecode.lib.dbmigrate.migrate.exceptions import MigrateDeprecationWarning, InvalidScriptError, ScriptError
 
 log = logging.getLogger(__name__)
 __all__ = ['PythonScript']
 
 
 class PythonScript(base.BaseScript):
     """Base for Python scripts"""
 
     @classmethod
     def create(cls, path, **opts):
         """Create an empty migration script at specified path
 
         :returns: :class:`PythonScript instance <migrate.versioning.script.py.PythonScript>`"""
         cls.require_notfound(path)
 
         src = Template(opts.pop('templates_path', None)).get_script(theme=opts.pop('templates_theme', None))
         shutil.copy(src, path)
 
         return cls(path)
 
     @classmethod
     def make_update_script_for_model(cls, engine, oldmodel,
                                      model, repository, **opts):
         """Create a migration script based on difference between two SA models.
 
         :param repository: path to migrate repository
         :param oldmodel: dotted.module.name:SAClass or SAClass object
         :param model: dotted.module.name:SAClass or SAClass object
         :param engine: SQLAlchemy engine
         :type repository: string or :class:`Repository instance <migrate.versioning.repository.Repository>`
         :type oldmodel: string or Class
         :type model: string or Class
         :type engine: Engine instance
         :returns: Upgrade / Downgrade script
         :rtype: string
         """
 
         if isinstance(repository, str):
             # oh dear, an import cycle!
             from rhodecode.lib.dbmigrate.migrate.versioning.repository import Repository
             repository = Repository(repository)
 
         oldmodel = load_model(oldmodel)
         model = load_model(model)
 
         # Compute differences.
         diff = schemadiff.getDiffOfModelAgainstModel(
             model,
             oldmodel,
             excludeTables=[repository.version_table])
         # TODO: diff can be False (there is no difference?)
         decls, upgradeCommands, downgradeCommands = \
             genmodel.ModelGenerator(diff,engine).genB2AMigration()
 
         # Store differences into file.
         src = Template(opts.pop('templates_path', None)).get_script(opts.pop('templates_theme', None))
         with open(src) as f:
             contents = f.read()
 
         # generate source
         search = 'def upgrade(migrate_engine):'
         contents = contents.replace(search, '\n\n'.join((decls, search)), 1)
         if upgradeCommands:
             contents = contents.replace('    pass', upgradeCommands, 1)
         if downgradeCommands:
             contents = contents.replace('    pass', downgradeCommands, 1)
         return contents
 
     @classmethod
     def verify_module(cls, path):
         """Ensure path is a valid script
 
         :param path: Script location
         :type path: string
         :raises: :exc:`InvalidScriptError <migrate.exceptions.InvalidScriptError>`
         :returns: Python module
         """
         # Try to import and get the upgrade() func
         module = import_path(path)
         try:
             assert callable(module.upgrade)
         except Exception as e:
             raise InvalidScriptError(path + ': %s' % str(e))
         return module
 
     def preview_sql(self, url, step, **args):
         """Mocks SQLAlchemy Engine to store all executed calls in a string
         and runs :meth:`PythonScript.run <migrate.versioning.script.py.PythonScript.run>`
 
         :returns: SQL file
         """
         buf = StringIO()
         args['engine_arg_strategy'] = 'mock'
         args['engine_arg_executor'] = lambda s, p = '': buf.write(str(s) + p)
 
         @with_engine
         def go(url, step, **kw):
             engine = kw.pop('engine')
             self.run(engine, step)
             return buf.getvalue()
 
         return go(url, step, **args)
 
     def run(self, engine, step):
         """Core method of Script file.
         Exectues :func:`update` or :func:`downgrade` functions
 
         :param engine: SQLAlchemy Engine
         :param step: Operation to run
         :type engine: string
         :type step: int
         """
         if step > 0:
             op = 'upgrade'
         elif step < 0:
             op = 'downgrade'
         else:
             raise ScriptError("%d is not a valid step" % step)
 
         funcname = base.operations[op]
         script_func = self._func(funcname)
 
         # check for old way of using engine
         if not inspect.getargspec(script_func)[0]:
             raise TypeError("upgrade/downgrade functions must accept engine"
                 " parameter (since version 0.5.4)")
 
         script_func(engine)
 
     @property
     def module(self):
         """Calls :meth:`migrate.versioning.script.py.verify_module`
         and returns it.
         """
         if not hasattr(self, '_module'):
             self._module = self.verify_module(self.path)
         return self._module
 
     def _func(self, funcname):
         if not hasattr(self.module, funcname):
             msg = "Function '%s' is not defined in this script"
             raise ScriptError(msg % funcname)
         return getattr(self.module, funcname)

rhodecode/lib/feedgenerator/feedgenerator.py

@@ -1,446 +1,446 b''
 # Copyright (c) Django Software Foundation and individual contributors.
 # All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without modification,
 # are permitted provided that the following conditions are met:
 #
 #     1. Redistributions of source code must retain the above copyright notice,
 #        this list of conditions and the following disclaimer.
 #
 #     2. Redistributions in binary form must reproduce the above copyright
 #        notice, this list of conditions and the following disclaimer in the
 #        documentation and/or other materials provided with the distribution.
 #
 #     3. Neither the name of Django nor the names of its contributors may be used
 #        to endorse or promote products derived from this software without
 #        specific prior written permission.
 #
 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
 # ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 # DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
 # ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 # (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 # LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
 # ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 """
 For definitions of the different versions of RSS, see:
 http://web.archive.org/web/20110718035220/http://diveintomark.org/archives/2004/02/04/incompatible-rss
 """
 
 
 import datetime
-from StringIO import StringIO
+from io import StringIO
 
 import pytz
 from six.moves.urllib import parse as urlparse
 
 from rhodecode.lib.feedgenerator import datetime_safe
 from rhodecode.lib.feedgenerator.utils import SimplerXMLGenerator, iri_to_uri, force_text
 
 
 #### The following code comes from ``django.utils.feedgenerator`` ####
 
 
 def rfc2822_date(date):
     # We can't use strftime() because it produces locale-dependent results, so
     # we have to map english month and day names manually
     months = ('Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec',)
     days = ('Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun')
     # Support datetime objects older than 1900
     date = datetime_safe.new_datetime(date)
     # We do this ourselves to be timezone aware, email.Utils is not tz aware.
     dow = days[date.weekday()]
     month = months[date.month - 1]
     time_str = date.strftime('%s, %%d %s %%Y %%H:%%M:%%S ' % (dow, month))
 
     time_str = time_str.decode('utf-8')
     offset = date.utcoffset()
     # Historically, this function assumes that naive datetimes are in UTC.
     if offset is None:
         return time_str + '-0000'
     else:
         timezone = (offset.days * 24 * 60) + (offset.seconds // 60)
         hour, minute = divmod(timezone, 60)
         return time_str + '%+03d%02d' % (hour, minute)
 
 
 def rfc3339_date(date):
     # Support datetime objects older than 1900
     date = datetime_safe.new_datetime(date)
     time_str = date.strftime('%Y-%m-%dT%H:%M:%S')
 
     time_str = time_str.decode('utf-8')
     offset = date.utcoffset()
     # Historically, this function assumes that naive datetimes are in UTC.
     if offset is None:
         return time_str + 'Z'
     else:
         timezone = (offset.days * 24 * 60) + (offset.seconds // 60)
         hour, minute = divmod(timezone, 60)
         return time_str + '%+03d:%02d' % (hour, minute)
 
 
 def get_tag_uri(url, date):
     """
     Creates a TagURI.
 
     See http://web.archive.org/web/20110514113830/http://diveintomark.org/archives/2004/05/28/howto-atom-id
     """
     bits = urlparse(url)
     d = ''
     if date is not None:
         d = ',%s' % datetime_safe.new_datetime(date).strftime('%Y-%m-%d')
     return 'tag:%s%s:%s/%s' % (bits.hostname, d, bits.path, bits.fragment)
 
 
 class SyndicationFeed(object):
     """Base class for all syndication feeds. Subclasses should provide write()"""
 
     def __init__(self, title, link, description, language=None, author_email=None,
                  author_name=None, author_link=None, subtitle=None, categories=None,
                  feed_url=None, feed_copyright=None, feed_guid=None, ttl=None, **kwargs):
         def to_unicode(s):
             return force_text(s, strings_only=True)
         if categories:
             categories = [force_text(c) for c in categories]
         if ttl is not None:
             # Force ints to unicode
             ttl = force_text(ttl)
         self.feed = {
             'title': to_unicode(title),
             'link': iri_to_uri(link),
             'description': to_unicode(description),
             'language': to_unicode(language),
             'author_email': to_unicode(author_email),
             'author_name': to_unicode(author_name),
             'author_link': iri_to_uri(author_link),
             'subtitle': to_unicode(subtitle),
             'categories': categories or (),
             'feed_url': iri_to_uri(feed_url),
             'feed_copyright': to_unicode(feed_copyright),
             'id': feed_guid or link,
             'ttl': ttl,
         }
         self.feed.update(kwargs)
         self.items = []
 
     def add_item(self, title, link, description, author_email=None,
                  author_name=None, author_link=None, pubdate=None, comments=None,
                  unique_id=None, unique_id_is_permalink=None, enclosure=None,
                  categories=(), item_copyright=None, ttl=None, updateddate=None,
                  enclosures=None, **kwargs):
         """
         Adds an item to the feed. All args are expected to be Python Unicode
         objects except pubdate and updateddate, which are datetime.datetime
         objects, and enclosures, which is an iterable of instances of the
         Enclosure class.
         """
         def to_unicode(s):
             return force_text(s, strings_only=True)
         if categories:
             categories = [to_unicode(c) for c in categories]
         if ttl is not None:
             # Force ints to unicode
             ttl = force_text(ttl)
         if enclosure is None:
             enclosures = [] if enclosures is None else enclosures
 
         item = {
             'title': to_unicode(title),
             'link': iri_to_uri(link),
             'description': to_unicode(description),
             'author_email': to_unicode(author_email),
             'author_name': to_unicode(author_name),
             'author_link': iri_to_uri(author_link),
             'pubdate': pubdate,
             'updateddate': updateddate,
             'comments': to_unicode(comments),
             'unique_id': to_unicode(unique_id),
             'unique_id_is_permalink': unique_id_is_permalink,
             'enclosures': enclosures,
             'categories': categories or (),
             'item_copyright': to_unicode(item_copyright),
             'ttl': ttl,
         }
         item.update(kwargs)
         self.items.append(item)
 
     def num_items(self):
         return len(self.items)
 
     def root_attributes(self):
         """
         Return extra attributes to place on the root (i.e. feed/channel) element.
         Called from write().
         """
         return {}
 
     def add_root_elements(self, handler):
         """
         Add elements in the root (i.e. feed/channel) element. Called
         from write().
         """
         pass
 
     def item_attributes(self, item):
         """
         Return extra attributes to place on each item (i.e. item/entry) element.
         """
         return {}
 
     def add_item_elements(self, handler, item):
         """
         Add elements on each item (i.e. item/entry) element.
         """
         pass
 
     def write(self, outfile, encoding):
         """
         Outputs the feed in the given encoding to outfile, which is a file-like
         object. Subclasses should override this.
         """
         raise NotImplementedError('subclasses of SyndicationFeed must provide a write() method')
 
     def writeString(self, encoding):
         """
         Returns the feed in the given encoding as a string.
         """
         s = StringIO()
         self.write(s, encoding)
         return s.getvalue()
 
     def latest_post_date(self):
         """
         Returns the latest item's pubdate or updateddate. If no items
         have either of these attributes this returns the current UTC date/time.
         """
         latest_date = None
         date_keys = ('updateddate', 'pubdate')
 
         for item in self.items:
             for date_key in date_keys:
                 item_date = item.get(date_key)
                 if item_date:
                     if latest_date is None or item_date > latest_date:
                         latest_date = item_date
 
         # datetime.now(tz=utc) is slower, as documented in django.utils.timezone.now
         return latest_date or datetime.datetime.utcnow().replace(tzinfo=pytz.utc)
 
 
 class Enclosure(object):
     """Represents an RSS enclosure"""
     def __init__(self, url, length, mime_type):
         """All args are expected to be Python Unicode objects"""
         self.length, self.mime_type = length, mime_type
         self.url = iri_to_uri(url)
 
 
 class RssFeed(SyndicationFeed):
     content_type = 'application/rss+xml; charset=utf-8'
 
     def write(self, outfile, encoding):
         handler = SimplerXMLGenerator(outfile, encoding)
         handler.startDocument()
         handler.startElement("rss", self.rss_attributes())
         handler.startElement("channel", self.root_attributes())
         self.add_root_elements(handler)
         self.write_items(handler)
         self.endChannelElement(handler)
         handler.endElement("rss")
 
     def rss_attributes(self):
         return {"version": self._version,
                 "xmlns:atom": "http://www.w3.org/2005/Atom"}
 
     def write_items(self, handler):
         for item in self.items:
             handler.startElement('item', self.item_attributes(item))
             self.add_item_elements(handler, item)
             handler.endElement("item")
 
     def add_root_elements(self, handler):
         handler.addQuickElement("title", self.feed['title'])
         handler.addQuickElement("link", self.feed['link'])
         handler.addQuickElement("description", self.feed['description'])
         if self.feed['feed_url'] is not None:
             handler.addQuickElement("atom:link", None, {"rel": "self", "href": self.feed['feed_url']})
         if self.feed['language'] is not None:
             handler.addQuickElement("language", self.feed['language'])
         for cat in self.feed['categories']:
             handler.addQuickElement("category", cat)
         if self.feed['feed_copyright'] is not None:
             handler.addQuickElement("copyright", self.feed['feed_copyright'])
         handler.addQuickElement("lastBuildDate", rfc2822_date(self.latest_post_date()))
         if self.feed['ttl'] is not None:
             handler.addQuickElement("ttl", self.feed['ttl'])
 
     def endChannelElement(self, handler):
         handler.endElement("channel")
 
 
 class RssUserland091Feed(RssFeed):
     _version = "0.91"
 
     def add_item_elements(self, handler, item):
         handler.addQuickElement("title", item['title'])
         handler.addQuickElement("link", item['link'])
         if item['description'] is not None:
             handler.addQuickElement("description", item['description'])
 
 
 class Rss201rev2Feed(RssFeed):
     # Spec: http://blogs.law.harvard.edu/tech/rss
     _version = "2.0"
 
     def add_item_elements(self, handler, item):
         handler.addQuickElement("title", item['title'])
         handler.addQuickElement("link", item['link'])
         if item['description'] is not None:
             handler.addQuickElement("description", item['description'])
 
         # Author information.
         if item["author_name"] and item["author_email"]:
             handler.addQuickElement("author", "%s (%s)" % (item['author_email'], item['author_name']))
         elif item["author_email"]:
             handler.addQuickElement("author", item["author_email"])
         elif item["author_name"]:
             handler.addQuickElement(
                 "dc:creator", item["author_name"], {"xmlns:dc": "http://purl.org/dc/elements/1.1/"}
             )
 
         if item['pubdate'] is not None:
             handler.addQuickElement("pubDate", rfc2822_date(item['pubdate']))
         if item['comments'] is not None:
             handler.addQuickElement("comments", item['comments'])
         if item['unique_id'] is not None:
             guid_attrs = {}
             if isinstance(item.get('unique_id_is_permalink'), bool):
                 guid_attrs['isPermaLink'] = str(item['unique_id_is_permalink']).lower()
             handler.addQuickElement("guid", item['unique_id'], guid_attrs)
         if item['ttl'] is not None:
             handler.addQuickElement("ttl", item['ttl'])
 
         # Enclosure.
         if item['enclosures']:
             enclosures = list(item['enclosures'])
             if len(enclosures) > 1:
                 raise ValueError(
                     "RSS feed items may only have one enclosure, see "
                     "http://www.rssboard.org/rss-profile#element-channel-item-enclosure"
                 )
             enclosure = enclosures[0]
             handler.addQuickElement('enclosure', '', {
                 'url': enclosure.url,
                 'length': enclosure.length,
                 'type': enclosure.mime_type,
             })
 
         # Categories.
         for cat in item['categories']:
             handler.addQuickElement("category", cat)
 
 
 class Atom1Feed(SyndicationFeed):
     # Spec: https://tools.ietf.org/html/rfc4287
     content_type = 'application/atom+xml; charset=utf-8'
     ns = "http://www.w3.org/2005/Atom"
 
     def write(self, outfile, encoding):
         handler = SimplerXMLGenerator(outfile, encoding)
         handler.startDocument()
         handler.startElement('feed', self.root_attributes())
         self.add_root_elements(handler)
         self.write_items(handler)
         handler.endElement("feed")
 
     def root_attributes(self):
         if self.feed['language'] is not None:
             return {"xmlns": self.ns, "xml:lang": self.feed['language']}
         else:
             return {"xmlns": self.ns}
 
     def add_root_elements(self, handler):
         handler.addQuickElement("title", self.feed['title'])
         handler.addQuickElement("link", "", {"rel": "alternate", "href": self.feed['link']})
         if self.feed['feed_url'] is not None:
             handler.addQuickElement("link", "", {"rel": "self", "href": self.feed['feed_url']})
         handler.addQuickElement("id", self.feed['id'])
         handler.addQuickElement("updated", rfc3339_date(self.latest_post_date()))
         if self.feed['author_name'] is not None:
             handler.startElement("author", {})
             handler.addQuickElement("name", self.feed['author_name'])
             if self.feed['author_email'] is not None:
                 handler.addQuickElement("email", self.feed['author_email'])
             if self.feed['author_link'] is not None:
                 handler.addQuickElement("uri", self.feed['author_link'])
             handler.endElement("author")
         if self.feed['subtitle'] is not None:
             handler.addQuickElement("subtitle", self.feed['subtitle'])
         for cat in self.feed['categories']:
             handler.addQuickElement("category", "", {"term": cat})
         if self.feed['feed_copyright'] is not None:
             handler.addQuickElement("rights", self.feed['feed_copyright'])
 
     def write_items(self, handler):
         for item in self.items:
             handler.startElement("entry", self.item_attributes(item))
             self.add_item_elements(handler, item)
             handler.endElement("entry")
 
     def add_item_elements(self, handler, item):
         handler.addQuickElement("title", item['title'])
         handler.addQuickElement("link", "", {"href": item['link'], "rel": "alternate"})
 
         if item['pubdate'] is not None:
             handler.addQuickElement('published', rfc3339_date(item['pubdate']))
 
         if item['updateddate'] is not None:
             handler.addQuickElement('updated', rfc3339_date(item['updateddate']))
 
         # Author information.
         if item['author_name'] is not None:
             handler.startElement("author", {})
             handler.addQuickElement("name", item['author_name'])
             if item['author_email'] is not None:
                 handler.addQuickElement("email", item['author_email'])
             if item['author_link'] is not None:
                 handler.addQuickElement("uri", item['author_link'])
             handler.endElement("author")
 
         # Unique ID.
         if item['unique_id'] is not None:
             unique_id = item['unique_id']
         else:
             unique_id = get_tag_uri(item['link'], item['pubdate'])
         handler.addQuickElement("id", unique_id)
 
         # Summary.
         if item['description'] is not None:
             handler.addQuickElement("summary", item['description'], {"type": "html"})
 
         # Enclosures.
         for enclosure in item['enclosures']:
             handler.addQuickElement('link', '', {
                 'rel': 'enclosure',
                 'href': enclosure.url,
                 'length': enclosure.length,
                 'type': enclosure.mime_type,
             })
 
         # Categories.
         for cat in item['categories']:
             handler.addQuickElement("category", "", {"term": cat})
 
         # Rights.
         if item['item_copyright'] is not None:
             handler.addQuickElement("rights", item['item_copyright'])
 
 
 # This isolates the decision of what the system default is, so calling code can
 # do "feedgenerator.DefaultFeed" instead of "feedgenerator.Rss201rev2Feed".
 DefaultFeed = Rss201rev2Feed
  No newline at end of file

rhodecode/lib/helpers.py

@@ -1,2155 +1,2155 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 Helper functions
 
 Consists of functions to typically be used within templates, but also
 available to Controllers. This module is available to both as 'h'.
 """
 import base64
 import collections
 
 import os
 import random
 import hashlib
-import StringIO
+from io import StringIO
 import textwrap
 import urllib.request, urllib.parse, urllib.error
 import math
 import logging
 import re
 import time
 import string
 import hashlib
 import regex
 from collections import OrderedDict
 
 import pygments
 import itertools
 import fnmatch
 import bleach
 
 from datetime import datetime
 from functools import partial
 from pygments.formatters.html import HtmlFormatter
 from pygments.lexers import (
     get_lexer_by_name, get_lexer_for_filename, get_lexer_for_mimetype)
 
 from pyramid.threadlocal import get_current_request
 from tempita import looper
 from webhelpers2.html import literal, HTML, escape
 from webhelpers2.html._autolink import _auto_link_urls
 from webhelpers2.html.tools import (
     button_to, highlight, js_obfuscate, strip_links, strip_tags)
 
 from webhelpers2.text import (
     chop_at, collapse, convert_accented_entities,
     convert_misc_entities, lchop, plural, rchop, remove_formatting,
     replace_whitespace, urlify, truncate, wrap_paragraphs)
 from webhelpers2.date import time_ago_in_words
 
 from webhelpers2.html.tags import (
     _input, NotGiven, _make_safe_id_component as safeid,
     form as insecure_form,
     auto_discovery_link, checkbox, end_form, file,
     hidden, image, javascript_link, link_to, link_to_if, link_to_unless, ol,
     select as raw_select, stylesheet_link, submit, text, password, textarea,
     ul, radio, Options)
 
 from webhelpers2.number import format_byte_size
 
 from rhodecode.lib.action_parser import action_parser
 from rhodecode.lib.pagination import Page, RepoPage, SqlPage
 from rhodecode.lib.ext_json import json
 from rhodecode.lib.utils import repo_name_slug, get_custom_lexer
 from rhodecode.lib.utils2 import (
     str2bool, safe_unicode, safe_str,
     get_commit_safe, datetime_to_time, time_to_datetime, time_to_utcdatetime,
     AttributeDict, safe_int, md5, md5_safe, get_host_info)
 from rhodecode.lib.markup_renderer import MarkupRenderer, relative_links
 from rhodecode.lib.vcs.exceptions import CommitDoesNotExistError
 from rhodecode.lib.vcs.backends.base import BaseChangeset, EmptyCommit
 from rhodecode.lib.vcs.conf.settings import ARCHIVE_SPECS
 from rhodecode.lib.index.search_utils import get_matching_line_offsets
 from rhodecode.config.conf import DATE_FORMAT, DATETIME_FORMAT
 from rhodecode.model.changeset_status import ChangesetStatusModel
 from rhodecode.model.db import Permission, User, Repository, UserApiKeys, FileStore
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.settings import IssueTrackerSettingsModel
 
 
 log = logging.getLogger(__name__)
 
 
 DEFAULT_USER = User.DEFAULT_USER
 DEFAULT_USER_EMAIL = User.DEFAULT_USER_EMAIL
 
 
 def asset(path, ver=None, **kwargs):
     """
     Helper to generate a static asset file path for rhodecode assets
 
     eg. h.asset('images/image.png', ver='3923')
 
     :param path: path of asset
     :param ver: optional version query param to append as ?ver=
     """
     request = get_current_request()
     query = {}
     query.update(kwargs)
     if ver:
         query = {'ver': ver}
     return request.static_path(
         'rhodecode:public/{}'.format(path), _query=query)
 
 
 default_html_escape_table = {
     ord('&'): u'&amp;',
     ord('<'): u'&lt;',
     ord('>'): u'&gt;',
     ord('"'): u'&quot;',
     ord("'"): u'&#39;',
 }
 
 
 def html_escape(text, html_escape_table=default_html_escape_table):
     """Produce entities within text."""
     return text.translate(html_escape_table)
 
 
 def chop_at_smart(s, sub, inclusive=False, suffix_if_chopped=None):
     """
     Truncate string ``s`` at the first occurrence of ``sub``.
 
     If ``inclusive`` is true, truncate just after ``sub`` rather than at it.
     """
     suffix_if_chopped = suffix_if_chopped or ''
     pos = s.find(sub)
     if pos == -1:
         return s
 
     if inclusive:
         pos += len(sub)
 
     chopped = s[:pos]
     left = s[pos:].strip()
 
     if left and suffix_if_chopped:
         chopped += suffix_if_chopped
 
     return chopped
 
 
 def shorter(text, size=20, prefix=False):
     postfix = '...'
     if len(text) > size:
         if prefix:
             # shorten in front
             return postfix + text[-(size - len(postfix)):]
         else:
             return text[:size - len(postfix)] + postfix
     return text
 
 
 def reset(name, value=None, id=NotGiven, type="reset", **attrs):
     """
     Reset button
     """
     return _input(type, name, value, id, attrs)
 
 
 def select(name, selected_values, options, id=NotGiven, **attrs):
 
     if isinstance(options, (list, tuple)):
         options_iter = options
         # Handle old value,label lists ... where value also can be value,label lists
         options = Options()
         for opt in options_iter:
             if isinstance(opt, tuple) and len(opt) == 2:
                 value, label = opt
             elif isinstance(opt, str):
                 value = label = opt
             else:
                 raise ValueError('invalid select option type %r' % type(opt))
 
             if isinstance(value, (list, tuple)):
                 option_group = options.add_optgroup(label)
                 for opt2 in value:
                     if isinstance(opt2, tuple) and len(opt2) == 2:
                         group_value, group_label = opt2
                     elif isinstance(opt2, str):
                         group_value = group_label = opt2
                     else:
                         raise ValueError('invalid select option type %r' % type(opt2))
 
                     option_group.add_option(group_label, group_value)
             else:
                 options.add_option(label, value)
 
     return raw_select(name, selected_values, options, id=id, **attrs)
 
 
 def branding(name, length=40):
     return truncate(name, length, indicator="")
 
 
 def FID(raw_id, path):
     """
     Creates a unique ID for filenode based on it's hash of path and commit
     it's safe to use in urls
 
     :param raw_id:
     :param path:
     """
 
     return 'c-%s-%s' % (short_id(raw_id), md5_safe(path)[:12])
 
 
 class _GetError(object):
     """Get error from form_errors, and represent it as span wrapped error
     message
 
     :param field_name: field to fetch errors for
     :param form_errors: form errors dict
     """
 
     def __call__(self, field_name, form_errors):
         tmpl = """<span class="error_msg">%s</span>"""
         if form_errors and field_name in form_errors:
             return literal(tmpl % form_errors.get(field_name))
 
 
 get_error = _GetError()
 
 
 class _ToolTip(object):
 
     def __call__(self, tooltip_title, trim_at=50):
         """
         Special function just to wrap our text into nice formatted
         autowrapped text
 
         :param tooltip_title:
         """
         tooltip_title = escape(tooltip_title)
         tooltip_title = tooltip_title.replace('<', '&lt;').replace('>', '&gt;')
         return tooltip_title
 
 
 tooltip = _ToolTip()
 
 files_icon = u'<i class="file-breadcrumb-copy tooltip icon-clipboard clipboard-action" data-clipboard-text="{}" title="Copy file path"></i>'
 
 
 def files_breadcrumbs(repo_name, repo_type, commit_id, file_path, landing_ref_name=None, at_ref=None,
                       limit_items=False, linkify_last_item=False, hide_last_item=False,
                       copy_path_icon=True):
     if isinstance(file_path, str):
         file_path = safe_unicode(file_path)
 
     if at_ref:
         route_qry = {'at': at_ref}
         default_landing_ref = at_ref or landing_ref_name or commit_id
     else:
         route_qry = None
         default_landing_ref = commit_id
 
     # first segment is a `HOME` link to repo files root location
     root_name = literal(u'<i class="icon-home"></i>')
 
     url_segments = [
         link_to(
             root_name,
             repo_files_by_ref_url(
                 repo_name,
                 repo_type,
                 f_path=None,  # None here is a special case for SVN repos,
                               # that won't prefix with a ref
                 ref_name=default_landing_ref,
                 commit_id=commit_id,
                 query=route_qry
             )
         )]
 
     path_segments = file_path.split('/')
     last_cnt = len(path_segments) - 1
     for cnt, segment in enumerate(path_segments):
         if not segment:
             continue
         segment_html = escape(segment)
 
         last_item = cnt == last_cnt
 
         if last_item and hide_last_item:
             # iterate over and hide last element
             continue
 
         if last_item and linkify_last_item is False:
             # plain version
             url_segments.append(segment_html)
         else:
             url_segments.append(
                 link_to(
                     segment_html,
                     repo_files_by_ref_url(
                         repo_name,
                         repo_type,
                         f_path='/'.join(path_segments[:cnt + 1]),
                         ref_name=default_landing_ref,
                         commit_id=commit_id,
                         query=route_qry
                     ),
                 ))
 
     limited_url_segments = url_segments[:1] + ['...'] + url_segments[-5:]
     if limit_items and len(limited_url_segments) < len(url_segments):
         url_segments = limited_url_segments
 
     full_path = file_path
     if copy_path_icon:
         icon = files_icon.format(escape(full_path))
     else:
         icon = ''
 
     if file_path == '':
         return root_name
     else:
         return literal(' / '.join(url_segments) + icon)
 
 
 def files_url_data(request):
     import urllib.request, urllib.parse, urllib.error
     matchdict = request.matchdict
 
     if 'f_path' not in matchdict:
         matchdict['f_path'] = ''
     else:
         matchdict['f_path'] = urllib.parse.quote(safe_str(matchdict['f_path']))
     if 'commit_id' not in matchdict:
         matchdict['commit_id'] = 'tip'
 
     return json.dumps(matchdict)
 
 
 def repo_files_by_ref_url(db_repo_name, db_repo_type, f_path, ref_name, commit_id, query=None, ):
     _is_svn = is_svn(db_repo_type)
     final_f_path = f_path
 
     if _is_svn:
         """
         For SVN the ref_name cannot be used as a commit_id, it needs to be prefixed with
         actually commit_id followed by the ref_name. This should be done only in case
         This is a initial landing url, without additional paths.
 
         like: /1000/tags/1.0.0/?at=tags/1.0.0
         """
 
         if ref_name and ref_name != 'tip':
             # NOTE(marcink): for svn the ref_name is actually the stored path, so we prefix it
             # for SVN we only do this magic prefix if it's root, .eg landing revision
             # of files link. If we are in the tree we don't need this since we traverse the url
             # that has everything stored
             if f_path in ['', '/']:
                 final_f_path = '/'.join([ref_name, f_path])
 
         # SVN always needs a commit_id explicitly, without a named REF
         default_commit_id = commit_id
     else:
         """
         For git and mercurial we construct a new URL using the names instead of commit_id
         like: /master/some_path?at=master
         """
         # We currently do not support branches with slashes
         if '/' in ref_name:
             default_commit_id = commit_id
         else:
             default_commit_id = ref_name
 
     # sometimes we pass f_path as None, to indicate explicit no prefix,
     # we translate it to string to not have None
     final_f_path = final_f_path or ''
 
     files_url = route_path(
         'repo_files',
         repo_name=db_repo_name,
         commit_id=default_commit_id,
         f_path=final_f_path,
         _query=query
     )
     return files_url
 
 
 def code_highlight(code, lexer, formatter, use_hl_filter=False):
     """
     Lex ``code`` with ``lexer`` and format it with the formatter ``formatter``.
 
     If ``outfile`` is given and a valid file object (an object
     with a ``write`` method), the result will be written to it, otherwise
     it is returned as a string.
     """
     if use_hl_filter:
         # add HL filter
         from rhodecode.lib.index import search_utils
         lexer.add_filter(search_utils.ElasticSearchHLFilter())
     return pygments.format(pygments.lex(code, lexer), formatter)
 
 
 class CodeHtmlFormatter(HtmlFormatter):
     """
     My code Html Formatter for source codes
     """
 
     def wrap(self, source, outfile):
         return self._wrap_div(self._wrap_pre(self._wrap_code(source)))
 
     def _wrap_code(self, source):
         for cnt, it in enumerate(source):
             i, t = it
             t = '<div id="L%s">%s</div>' % (cnt + 1, t)
             yield i, t
 
     def _wrap_tablelinenos(self, inner):
         dummyoutfile = StringIO.StringIO()
         lncount = 0
         for t, line in inner:
             if t:
                 lncount += 1
             dummyoutfile.write(line)
 
         fl = self.linenostart
         mw = len(str(lncount + fl - 1))
         sp = self.linenospecial
         st = self.linenostep
         la = self.lineanchors
         aln = self.anchorlinenos
         nocls = self.noclasses
         if sp:
             lines = []
 
             for i in range(fl, fl + lncount):
                 if i % st == 0:
                     if i % sp == 0:
                         if aln:
                             lines.append('<a href="#%s%d" class="special">%*d</a>' %
                                          (la, i, mw, i))
                         else:
                             lines.append('<span class="special">%*d</span>' % (mw, i))
                     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" '
                       'style="background-color: #f0f0f0; padding-right: 10px">'
                       '<pre style="line-height: 125%">' +
                       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, dummyoutfile.getvalue()
         yield 0, '</td></tr></table>'
 
 
 class SearchContentCodeHtmlFormatter(CodeHtmlFormatter):
     def __init__(self, **kw):
         # only show these line numbers if set
         self.only_lines = kw.pop('only_line_numbers', [])
         self.query_terms = kw.pop('query_terms', [])
         self.max_lines = kw.pop('max_lines', 5)
         self.line_context = kw.pop('line_context', 3)
         self.url = kw.pop('url', None)
 
         super(CodeHtmlFormatter, self).__init__(**kw)
 
     def _wrap_code(self, source):
         for cnt, it in enumerate(source):
             i, t = it
             t = '<pre>%s</pre>' % t
             yield i, t
 
     def _wrap_tablelinenos(self, inner):
         yield 0, '<table class="code-highlight %stable">' % self.cssclass
 
         last_shown_line_number = 0
         current_line_number = 1
 
         for t, line in inner:
             if not t:
                 yield t, line
                 continue
 
             if current_line_number in self.only_lines:
                 if last_shown_line_number + 1 != current_line_number:
                     yield 0, '<tr>'
                     yield 0, '<td class="line">...</td>'
                     yield 0, '<td id="hlcode" class="code"></td>'
                     yield 0, '</tr>'
 
                 yield 0, '<tr>'
                 if self.url:
                     yield 0, '<td class="line"><a href="%s#L%i">%i</a></td>' % (
                         self.url, current_line_number, current_line_number)
                 else:
                     yield 0, '<td class="line"><a href="">%i</a></td>' % (
                         current_line_number)
                 yield 0, '<td id="hlcode" class="code">' + line + '</td>'
                 yield 0, '</tr>'
 
                 last_shown_line_number = current_line_number
 
             current_line_number += 1
 
         yield 0, '</table>'
 
 
 def hsv_to_rgb(h, s, v):
     """ Convert hsv color values to rgb """
 
     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 unique_color_generator(n=10000, saturation=0.10, lightness=0.95):
     """
     Generator for getting n of evenly distributed colors using
     hsv color and golden ratio. It always return same order of colors
 
     :param n: number of colors to generate
     :param saturation: saturation of returned colors
     :param lightness: lightness of returned colors
     :returns: RGB tuple
     """
 
     golden_ratio = 0.618033988749895
     h = 0.22717784590367374
 
     for _ in range(n):
         h += golden_ratio
         h %= 1
         HSV_tuple = [h, saturation, lightness]
         RGB_tuple = hsv_to_rgb(*HSV_tuple)
         yield map(lambda x: str(int(x * 256)), RGB_tuple)
 
 
 def color_hasher(n=10000, saturation=0.10, lightness=0.95):
     """
     Returns a function which when called with an argument returns a unique
     color for that argument, eg.
 
     :param n: number of colors to generate
     :param saturation: saturation of returned colors
     :param lightness: lightness of returned colors
     :returns: css RGB string
 
     >>> color_hash = color_hasher()
     >>> color_hash('hello')
     'rgb(34, 12, 59)'
     >>> color_hash('hello')
     'rgb(34, 12, 59)'
     >>> color_hash('other')
     'rgb(90, 224, 159)'
     """
 
     color_dict = {}
     cgenerator = unique_color_generator(
         saturation=saturation, lightness=lightness)
 
     def get_color_string(thing):
         if thing in color_dict:
             col = color_dict[thing]
         else:
             col = color_dict[thing] = cgenerator.next()
         return "rgb(%s)" % (', '.join(col))
 
     return get_color_string
 
 
 def get_lexer_safe(mimetype=None, filepath=None):
     """
     Tries to return a relevant pygments lexer using mimetype/filepath name,
     defaulting to plain text if none could be found
     """
     lexer = None
     try:
         if mimetype:
             lexer = get_lexer_for_mimetype(mimetype)
         if not lexer:
             lexer = get_lexer_for_filename(filepath)
     except pygments.util.ClassNotFound:
         pass
 
     if not lexer:
         lexer = get_lexer_by_name('text')
 
     return lexer
 
 
 def get_lexer_for_filenode(filenode):
     lexer = get_custom_lexer(filenode.extension) or filenode.lexer
     return lexer
 
 
 def pygmentize(filenode, **kwargs):
     """
     pygmentize function using pygments
 
     :param filenode:
     """
     lexer = get_lexer_for_filenode(filenode)
     return literal(code_highlight(filenode.content, lexer,
                                   CodeHtmlFormatter(**kwargs)))
 
 
 def is_following_repo(repo_name, user_id):
     from rhodecode.model.scm import ScmModel
     return ScmModel().is_following_repo(repo_name, user_id)
 
 
 class _Message(object):
     """A message returned by ``Flash.pop_messages()``.
 
     Converting the message to a string returns the message text. Instances
     also have the following attributes:
 
     * ``message``: the message text.
     * ``category``: the category specified when the message was created.
     """
 
     def __init__(self, category, message, sub_data=None):
         self.category = category
         self.message = message
         self.sub_data = sub_data or {}
 
     def __str__(self):
         return self.message
 
     __unicode__ = __str__
 
     def __html__(self):
         return escape(safe_unicode(self.message))
 
 
 class Flash(object):
     # List of allowed categories.  If None, allow any category.
     categories = ["warning", "notice", "error", "success"]
 
     # Default category if none is specified.
     default_category = "notice"
 
     def __init__(self, session_key="flash", categories=None,
                  default_category=None):
         """
         Instantiate a ``Flash`` object.
 
         ``session_key`` is the key to save the messages under in the user's
         session.
 
         ``categories`` is an optional list which overrides the default list
         of categories.
 
         ``default_category`` overrides the default category used for messages
         when none is specified.
         """
         self.session_key = session_key
         if categories is not None:
             self.categories = categories
         if default_category is not None:
             self.default_category = default_category
         if self.categories and self.default_category not in self.categories:
             raise ValueError(
                 "unrecognized default category %r" % (self.default_category,))
 
     def pop_messages(self, session=None, request=None):
         """
         Return all accumulated messages and delete them from the session.
 
         The return value is a list of ``Message`` objects.
         """
         messages = []
 
         if not session:
             if not request:
                 request = get_current_request()
             session = request.session
 
         # Pop the 'old' pylons flash messages. They are tuples of the form
         # (category, message)
         for cat, msg in session.pop(self.session_key, []):
             messages.append(_Message(cat, msg))
 
         # Pop the 'new' pyramid flash messages for each category as list
         # of strings.
         for cat in self.categories:
             for msg in session.pop_flash(queue=cat):
                 sub_data = {}
                 if hasattr(msg, 'rsplit'):
                     flash_data = msg.rsplit('|DELIM|', 1)
                     org_message = flash_data[0]
                     if len(flash_data) > 1:
                         sub_data = json.loads(flash_data[1])
                 else:
                     org_message = msg
 
                 messages.append(_Message(cat, org_message, sub_data=sub_data))
 
         # Map messages from the default queue to the 'notice' category.
         for msg in session.pop_flash():
             messages.append(_Message('notice', msg))
 
         session.save()
         return messages
 
     def json_alerts(self, session=None, request=None):
         payloads = []
         messages = flash.pop_messages(session=session, request=request) or []
         for message in messages:
             payloads.append({
                 'message': {
                     'message': u'{}'.format(message.message),
                     'level': message.category,
                     'force': True,
                     'subdata': message.sub_data
                 }
             })
         return json.dumps(payloads)
 
     def __call__(self, message, category=None, ignore_duplicate=True,
                  session=None, request=None):
 
         if not session:
             if not request:
                 request = get_current_request()
             session = request.session
 
         session.flash(
             message, queue=category, allow_duplicate=not ignore_duplicate)
 
 
 flash = Flash()
 
 #==============================================================================
 # SCM FILTERS available via h.
 #==============================================================================
 from rhodecode.lib.vcs.utils import author_name, author_email
 from rhodecode.lib.utils2 import age, age_from_seconds
 from rhodecode.model.db import User, ChangesetStatus
 
 
 email = author_email
 
 
 def capitalize(raw_text):
     return raw_text.capitalize()
 
 
 def short_id(long_id):
     return long_id[:12]
 
 
 def hide_credentials(url):
     from rhodecode.lib.utils2 import credentials_filter
     return credentials_filter(url)
 
 
 import pytz
 import tzlocal
 local_timezone = tzlocal.get_localzone()
 
 
 def get_timezone(datetime_iso, time_is_local=False):
     tzinfo = '+00:00'
 
     # detect if we have a timezone info, otherwise, add it
     if time_is_local and isinstance(datetime_iso, datetime) and not datetime_iso.tzinfo:
         force_timezone = os.environ.get('RC_TIMEZONE', '')
         if force_timezone:
             force_timezone = pytz.timezone(force_timezone)
         timezone = force_timezone or local_timezone
         offset = timezone.localize(datetime_iso).strftime('%z')
         tzinfo = '{}:{}'.format(offset[:-2], offset[-2:])
     return tzinfo
 
 
 def age_component(datetime_iso, value=None, time_is_local=False, tooltip=True):
     title = value or format_date(datetime_iso)
     tzinfo = get_timezone(datetime_iso, time_is_local=time_is_local)
 
     return literal(
         '<time class="timeago {cls}" title="{tt_title}" datetime="{dt}{tzinfo}">{title}</time>'.format(
             cls='tooltip' if tooltip else '',
             tt_title=('{title}{tzinfo}'.format(title=title, tzinfo=tzinfo)) if tooltip else '',
             title=title, dt=datetime_iso, tzinfo=tzinfo
         ))
 
 
 def _shorten_commit_id(commit_id, commit_len=None):
     if commit_len is None:
         request = get_current_request()
         commit_len = request.call_context.visual.show_sha_length
     return commit_id[:commit_len]
 
 
 def show_id(commit, show_idx=None, commit_len=None):
     """
     Configurable function that shows ID
     by default it's r123:fffeeefffeee
 
     :param commit: commit instance
     """
     if show_idx is None:
         request = get_current_request()
         show_idx = request.call_context.visual.show_revision_number
 
     raw_id = _shorten_commit_id(commit.raw_id, commit_len=commit_len)
     if show_idx:
         return 'r%s:%s' % (commit.idx, raw_id)
     else:
         return '%s' % (raw_id, )
 
 
 def format_date(date):
     """
     use a standardized formatting for dates used in RhodeCode
 
     :param date: date/datetime object
     :return: formatted date
     """
 
     if date:
         _fmt = "%a, %d %b %Y %H:%M:%S"
         return safe_unicode(date.strftime(_fmt))
 
     return u""
 
 
 class _RepoChecker(object):
 
     def __init__(self, backend_alias):
         self._backend_alias = backend_alias
 
     def __call__(self, repository):
         if hasattr(repository, 'alias'):
             _type = repository.alias
         elif hasattr(repository, 'repo_type'):
             _type = repository.repo_type
         else:
             _type = repository
         return _type == self._backend_alias
 
 
 is_git = _RepoChecker('git')
 is_hg = _RepoChecker('hg')
 is_svn = _RepoChecker('svn')
 
 
 def get_repo_type_by_name(repo_name):
     repo = Repository.get_by_repo_name(repo_name)
     if repo:
         return repo.repo_type
 
 
 def is_svn_without_proxy(repository):
     if is_svn(repository):
         from rhodecode.model.settings import VcsSettingsModel
         conf = VcsSettingsModel().get_ui_settings_as_config_obj()
         return not str2bool(conf.get('vcs_svn_proxy', 'http_requests_enabled'))
     return False
 
 
 def discover_user(author):
     """
     Tries to discover RhodeCode User based on the author string. Author string
     is typically `FirstName LastName <[e-mail]>`
     """
 
     # if author is already an instance use it for extraction
     if isinstance(author, User):
         return author
 
     # Valid email in the attribute passed, see if they're in the system
     _email = author_email(author)
     if _email != '':
         user = User.get_by_email(_email, case_insensitive=True, cache=True)
         if user is not None:
             return user
 
     # Maybe it's a username, we try to extract it and fetch by username ?
     _author = author_name(author)
     user = User.get_by_username(_author, case_insensitive=True, cache=True)
     if user is not None:
         return user
 
     return None
 
 
 def email_or_none(author):
     # extract email from the commit string
     _email = author_email(author)
 
     # If we have an email, use it, otherwise
     # see if it contains a username we can get an email from
     if _email != '':
         return _email
     else:
         user = User.get_by_username(
             author_name(author), case_insensitive=True, cache=True)
 
     if user is not None:
             return user.email
 
     # No valid email, not a valid user in the system, none!
     return None
 
 
 def link_to_user(author, length=0, **kwargs):
     user = discover_user(author)
     # user can be None, but if we have it already it means we can re-use it
     # in the person() function, so we save 1 intensive-query
     if user:
         author = user
 
     display_person = person(author, 'username_or_name_or_email')
     if length:
         display_person = shorter(display_person, length)
 
     if user and user.username != user.DEFAULT_USER:
         return link_to(
             escape(display_person),
             route_path('user_profile', username=user.username),
             **kwargs)
     else:
         return escape(display_person)
 
 
 def link_to_group(users_group_name, **kwargs):
     return link_to(
         escape(users_group_name),
         route_path('user_group_profile', user_group_name=users_group_name),
         **kwargs)
 
 
 def person(author, show_attr="username_and_name"):
     user = discover_user(author)
     if user:
         return getattr(user, show_attr)
     else:
         _author = author_name(author)
         _email = email(author)
         return _author or _email
 
 
 def author_string(email):
     if email:
         user = User.get_by_email(email, case_insensitive=True, cache=True)
         if user:
             if user.first_name or user.last_name:
                 return '%s %s &lt;%s&gt;' % (
                     user.first_name, user.last_name, email)
             else:
                 return email
         else:
             return email
     else:
         return None
 
 
 def person_by_id(id_, show_attr="username_and_name"):
     # 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 gravatar_with_user(request, author, show_disabled=False, tooltip=False):
     _render = request.get_partial_renderer('rhodecode:templates/base/base.mako')
     return _render('gravatar_with_user', author, show_disabled=show_disabled, tooltip=tooltip)
 
 
 tags_paterns = OrderedDict((
     ('lang', (re.compile(r'\[(lang|language)\ \=\&gt;\ *([a-zA-Z\-\/\#\+\.]*)\]'),
              '<div class="metatag" tag="lang">\\2</div>')),
 
     ('see', (re.compile(r'\[see\ \=\&gt;\ *([a-zA-Z0-9\/\=\?\&amp;\ \:\/\.\-]*)\]'),
             '<div class="metatag" tag="see">see: \\1 </div>')),
 
     ('url', (re.compile(r'\[url\ \=\&gt;\ \[([a-zA-Z0-9\ \.\-\_]+)\]\((http://|https://|/)(.*?)\)\]'),
             '<div class="metatag" tag="url"> <a href="\\2\\3">\\1</a> </div>')),
 
     ('license', (re.compile(r'\[license\ \=\&gt;\ *([a-zA-Z0-9\/\=\?\&amp;\ \:\/\.\-]*)\]'),
                 '<div class="metatag" tag="license"><a href="http:\/\/www.opensource.org/licenses/\\1">\\1</a></div>')),
 
     ('ref', (re.compile(r'\[(requires|recommends|conflicts|base)\ \=\&gt;\ *([a-zA-Z0-9\-\/]*)\]'),
             '<div class="metatag" tag="ref \\1">\\1: <a href="/\\2">\\2</a></div>')),
 
     ('state', (re.compile(r'\[(stable|featured|stale|dead|dev|deprecated)\]'),
               '<div class="metatag" tag="state \\1">\\1</div>')),
 
     # label in grey
     ('label', (re.compile(r'\[([a-z]+)\]'),
               '<div class="metatag" tag="label">\\1</div>')),
 
     # generic catch all in grey
     ('generic', (re.compile(r'\[([a-zA-Z0-9\.\-\_]+)\]'),
                 '<div class="metatag" tag="generic">\\1</div>')),
 ))
 
 
 def extract_metatags(value):
     """
     Extract supported meta-tags from given text value
     """
     tags = []
     if not value:
         return tags, ''
 
     for key, val in tags_paterns.items():
         pat, replace_html = val
         tags.extend([(key, x.group()) for x in pat.finditer(value)])
         value = pat.sub('', value)
 
     return tags, value
 
 
 def style_metatag(tag_type, value):
     """
     converts tags from value into html equivalent
     """
     if not value:
         return ''
 
     html_value = value
     tag_data = tags_paterns.get(tag_type)
     if tag_data:
         pat, replace_html = tag_data
         # convert to plain `unicode` instead of a markup tag to be used in
         # regex expressions. safe_unicode doesn't work here
         html_value = pat.sub(replace_html, unicode(value))
 
     return html_value
 
 
 def bool2icon(value, show_at_false=True):
     """
     Returns boolean value of a given value, represented as html element with
     classes that will represent icons
 
     :param value: given value to convert to html node
     """
 
     if value:  # does bool conversion
         return HTML.tag('i', class_="icon-true", title='True')
     else:  # not true as bool
         if show_at_false:
             return HTML.tag('i', class_="icon-false", title='False')
         return HTML.tag('i')
 
 
 def b64(inp):
     return base64.b64encode(inp)
 
 #==============================================================================
 # PERMS
 #==============================================================================
 from rhodecode.lib.auth import (
     HasPermissionAny, HasPermissionAll,
     HasRepoPermissionAny, HasRepoPermissionAll, HasRepoGroupPermissionAll,
     HasRepoGroupPermissionAny, HasRepoPermissionAnyApi, get_csrf_token,
     csrf_token_key, AuthUser)
 
 
 #==============================================================================
 # GRAVATAR URL
 #==============================================================================
 class InitialsGravatar(object):
     def __init__(self, email_address, first_name, last_name, size=30,
                  background=None, text_color='#fff'):
         self.size = size
         self.first_name = first_name
         self.last_name = last_name
         self.email_address = email_address
         self.background = background or self.str2color(email_address)
         self.text_color = text_color
 
     def get_color_bank(self):
         """
         returns a predefined list of colors that gravatars can use.
         Those are randomized distinct colors that guarantee readability and
         uniqueness.
 
         generated with: http://phrogz.net/css/distinct-colors.html
         """
         return [
             '#bf3030', '#a67f53', '#00ff00', '#5989b3', '#392040', '#d90000',
             '#402910', '#204020', '#79baf2', '#a700b3', '#bf6060', '#7f5320',
             '#008000', '#003059', '#ee00ff', '#ff0000', '#8c4b00', '#007300',
             '#005fb3', '#de73e6', '#ff4040', '#ffaa00', '#3df255', '#203140',
             '#47004d', '#591616', '#664400', '#59b365', '#0d2133', '#83008c',
             '#592d2d', '#bf9f60', '#73e682', '#1d3f73', '#73006b', '#402020',
             '#b2862d', '#397341', '#597db3', '#e600d6', '#a60000', '#736039',
             '#00b318', '#79aaf2', '#330d30', '#ff8080', '#403010', '#16591f',
             '#002459', '#8c4688', '#e50000', '#ffbf40', '#00732e', '#102340',
             '#bf60ac', '#8c4646', '#cc8800', '#00a642', '#1d3473', '#b32d98',
             '#660e00', '#ffd580', '#80ffb2', '#7391e6', '#733967', '#d97b6c',
             '#8c5e00', '#59b389', '#3967e6', '#590047', '#73281d', '#665200',
             '#00e67a', '#2d50b3', '#8c2377', '#734139', '#b2982d', '#16593a',
             '#001859', '#ff00aa', '#a65e53', '#ffcc00', '#0d3321', '#2d3959',
             '#731d56', '#401610', '#4c3d00', '#468c6c', '#002ca6', '#d936a3',
             '#d94c36', '#403920', '#36d9a3', '#0d1733', '#592d4a', '#993626',
             '#cca300', '#00734d', '#46598c', '#8c005e', '#7f1100', '#8c7000',
             '#00a66f', '#7382e6', '#b32d74', '#d9896c', '#ffe680', '#1d7362',
             '#364cd9', '#73003d', '#d93a00', '#998a4d', '#59b3a1', '#5965b3',
             '#e5007a', '#73341d', '#665f00', '#00b38f', '#0018b3', '#59163a',
             '#b2502d', '#bfb960', '#00ffcc', '#23318c', '#a6537f', '#734939',
             '#b2a700', '#104036', '#3d3df2', '#402031', '#e56739', '#736f39',
             '#79f2ea', '#000059', '#401029', '#4c1400', '#ffee00', '#005953',
             '#101040', '#990052', '#402820', '#403d10', '#00ffee', '#0000d9',
             '#ff80c4', '#a66953', '#eeff00', '#00ccbe', '#8080ff', '#e673a1',
             '#a62c00', '#474d00', '#1a3331', '#46468c', '#733950', '#662900',
             '#858c23', '#238c85', '#0f0073', '#b20047', '#d9986c', '#becc00',
             '#396f73', '#281d73', '#ff0066', '#ff6600', '#dee673', '#59adb3',
             '#6559b3', '#590024', '#b2622d', '#98b32d', '#36ced9', '#332d59',
             '#40001a', '#733f1d', '#526600', '#005359', '#242040', '#bf6079',
             '#735039', '#cef23d', '#007780', '#5630bf', '#66001b', '#b24700',
             '#acbf60', '#1d6273', '#25008c', '#731d34', '#a67453', '#50592d',
             '#00ccff', '#6600ff', '#ff0044', '#4c1f00', '#8a994d', '#79daf2',
             '#a173e6', '#d93662', '#402310', '#aaff00', '#2d98b3', '#8c40ff',
             '#592d39', '#ff8c40', '#354020', '#103640', '#1a0040', '#331a20',
             '#331400', '#334d00', '#1d5673', '#583973', '#7f0022', '#4c3626',
             '#88cc00', '#36a3d9', '#3d0073', '#d9364c', '#33241a', '#698c23',
             '#5995b3', '#300059', '#e57382', '#7f3300', '#366600', '#00aaff',
             '#3a1659', '#733941', '#663600', '#74b32d', '#003c59', '#7f53a6',
             '#73000f', '#ff8800', '#baf279', '#79caf2', '#291040', '#a6293a',
             '#b2742d', '#587339', '#0077b3', '#632699', '#400009', '#d9a66c',
             '#294010', '#2d4a59', '#aa00ff', '#4c131b', '#b25f00', '#5ce600',
             '#267399', '#a336d9', '#990014', '#664e33', '#86bf60', '#0088ff',
             '#7700b3', '#593a16', '#073300', '#1d4b73', '#ac60bf', '#e59539',
             '#4f8c46', '#368dd9', '#5c0073'
         ]
 
     def rgb_to_hex_color(self, rgb_tuple):
         """
         Converts an rgb_tuple passed to an hex color.
 
         :param rgb_tuple: tuple with 3 ints represents rgb color space
         """
         return '#' + ("".join(map(chr, rgb_tuple)).encode('hex'))
 
     def email_to_int_list(self, email_str):
         """
         Get every byte of the hex digest value of email and turn it to integer.
         It's going to be always between 0-255
         """
         digest = md5_safe(email_str.lower())
         return [int(digest[i * 2:i * 2 + 2], 16) for i in range(16)]
 
     def pick_color_bank_index(self, email_str, color_bank):
         return self.email_to_int_list(email_str)[0] % len(color_bank)
 
     def str2color(self, email_str):
         """
         Tries to map in a stable algorithm an email to color
 
         :param email_str:
         """
         color_bank = self.get_color_bank()
         # pick position (module it's length so we always find it in the
         # bank even if it's smaller than 256 values
         pos = self.pick_color_bank_index(email_str, color_bank)
         return color_bank[pos]
 
     def normalize_email(self, email_address):
         import unicodedata
         # default host used to fill in the fake/missing email
         default_host = u'localhost'
 
         if not email_address:
             email_address = u'%s@%s' % (User.DEFAULT_USER, default_host)
 
         email_address = safe_unicode(email_address)
 
         if u'@' not in email_address:
             email_address = u'%s@%s' % (email_address, default_host)
 
         if email_address.endswith(u'@'):
             email_address = u'%s%s' % (email_address, default_host)
 
         email_address = unicodedata.normalize('NFKD', email_address)\
             .encode('ascii', 'ignore')
         return email_address
 
     def get_initials(self):
         """
         Returns 2 letter initials calculated based on the input.
         The algorithm picks first given email address, and takes first letter
         of part before @, and then the first letter of server name. In case
         the part before @ is in a format of `somestring.somestring2` it replaces
         the server letter with first letter of somestring2
 
         In case function was initialized with both first and lastname, this
         overrides the extraction from email by first letter of the first and
         last name. We add special logic to that functionality, In case Full name
         is compound, like Guido Von Rossum, we use last part of the last name
         (Von Rossum) picking `R`.
 
         Function also normalizes the non-ascii characters to they ascii
         representation, eg Ą => A
         """
         import unicodedata
         # replace non-ascii to ascii
         first_name = unicodedata.normalize(
             'NFKD', safe_unicode(self.first_name)).encode('ascii', 'ignore')
         last_name = unicodedata.normalize(
             'NFKD', safe_unicode(self.last_name)).encode('ascii', 'ignore')
 
         # do NFKD encoding, and also make sure email has proper format
         email_address = self.normalize_email(self.email_address)
 
         # first push the email initials
         prefix, server = email_address.split('@', 1)
 
         # check if prefix is maybe a 'first_name.last_name' syntax
         _dot_split = prefix.rsplit('.', 1)
         if len(_dot_split) == 2 and _dot_split[1]:
             initials = [_dot_split[0][0], _dot_split[1][0]]
         else:
             initials = [prefix[0], server[0]]
 
         # then try to replace either first_name or last_name
         fn_letter = (first_name or " ")[0].strip()
         ln_letter = (last_name.split(' ', 1)[-1] or " ")[0].strip()
 
         if fn_letter:
             initials[0] = fn_letter
 
         if ln_letter:
             initials[1] = ln_letter
 
         return ''.join(initials).upper()
 
     def get_img_data_by_type(self, font_family, img_type):
         default_user = """
         <svg xmlns="http://www.w3.org/2000/svg"
         version="1.1" x="0px" y="0px" width="{size}" height="{size}"
         viewBox="-15 -10 439.165 429.164"
 
         xml:space="preserve"
         style="background:{background};" >
 
         <path d="M204.583,216.671c50.664,0,91.74-48.075,
                  91.74-107.378c0-82.237-41.074-107.377-91.74-107.377
                  c-50.668,0-91.74,25.14-91.74,107.377C112.844,
                  168.596,153.916,216.671,
                  204.583,216.671z" fill="{text_color}"/>
         <path d="M407.164,374.717L360.88,
                  270.454c-2.117-4.771-5.836-8.728-10.465-11.138l-71.83-37.392
                  c-1.584-0.823-3.502-0.663-4.926,0.415c-20.316,
                  15.366-44.203,23.488-69.076,23.488c-24.877,
                  0-48.762-8.122-69.078-23.488
                  c-1.428-1.078-3.346-1.238-4.93-0.415L58.75,
                  259.316c-4.631,2.41-8.346,6.365-10.465,11.138L2.001,374.717
                  c-3.191,7.188-2.537,15.412,1.75,22.005c4.285,
                  6.592,11.537,10.526,19.4,10.526h362.861c7.863,0,15.117-3.936,
                  19.402-10.527 C409.699,390.129,
                  410.355,381.902,407.164,374.717z" fill="{text_color}"/>
         </svg>""".format(
             size=self.size,
             background='#979797',  # @grey4
             text_color=self.text_color,
             font_family=font_family)
 
         return {
             "default_user": default_user
         }[img_type]
 
     def get_img_data(self, svg_type=None):
         """
         generates the svg metadata for image
         """
         fonts = [
             '-apple-system',
              'BlinkMacSystemFont',
              'Segoe UI',
              'Roboto',
              'Oxygen-Sans',
              'Ubuntu',
              'Cantarell',
              'Helvetica Neue',
              'sans-serif'
         ]
         font_family = ','.join(fonts)
         if svg_type:
             return self.get_img_data_by_type(font_family, svg_type)
 
         initials = self.get_initials()
         img_data = """
         <svg xmlns="http://www.w3.org/2000/svg" pointer-events="none"
              width="{size}" height="{size}"
              style="width: 100%; height: 100%; background-color: {background}"
              viewBox="0 0 {size} {size}">
             <text text-anchor="middle" y="50%" x="50%" dy="0.35em"
                   pointer-events="auto" fill="{text_color}"
                   font-family="{font_family}"
                   style="font-weight: 400; font-size: {f_size}px;">{text}
             </text>
         </svg>""".format(
             size=self.size,
             f_size=self.size/2.05,  # scale the text inside the box nicely
             background=self.background,
             text_color=self.text_color,
             text=initials.upper(),
             font_family=font_family)
 
         return img_data
 
     def generate_svg(self, svg_type=None):
         img_data = self.get_img_data(svg_type)
         return "data:image/svg+xml;base64,%s" % base64.b64encode(img_data)
 
 
 def initials_gravatar(request, email_address, first_name, last_name, size=30, store_on_disk=False):
 
     svg_type = None
     if email_address == User.DEFAULT_USER_EMAIL:
         svg_type = 'default_user'
 
     klass = InitialsGravatar(email_address, first_name, last_name, size)
 
     if store_on_disk:
         from rhodecode.apps.file_store import utils as store_utils
         from rhodecode.apps.file_store.exceptions import FileNotAllowedException, \
             FileOverSizeException
         from rhodecode.model.db import Session
 
         image_key = md5_safe(email_address.lower()
                              + first_name.lower() + last_name.lower())
 
         storage = store_utils.get_file_storage(request.registry.settings)
         filename = '{}.svg'.format(image_key)
         subdir = 'gravatars'
         # since final name has a counter, we apply the 0
         uid = storage.apply_counter(0, store_utils.uid_filename(filename, randomized=False))
         store_uid = os.path.join(subdir, uid)
 
         db_entry = FileStore.get_by_store_uid(store_uid)
         if db_entry:
             return request.route_path('download_file', fid=store_uid)
 
         img_data = klass.get_img_data(svg_type=svg_type)
         img_file = store_utils.bytes_to_file_obj(img_data)
 
         try:
             store_uid, metadata = storage.save_file(
                 img_file, filename, directory=subdir,
                 extensions=['.svg'], randomized_name=False)
         except (FileNotAllowedException, FileOverSizeException):
             raise
 
         try:
             entry = FileStore.create(
                 file_uid=store_uid, filename=metadata["filename"],
                 file_hash=metadata["sha256"], file_size=metadata["size"],
                 file_display_name=filename,
                 file_description=u'user gravatar `{}`'.format(safe_unicode(filename)),
                 hidden=True, check_acl=False, user_id=1
             )
             Session().add(entry)
             Session().commit()
             log.debug('Stored upload in DB as %s', entry)
         except Exception:
             raise
 
         return request.route_path('download_file', fid=store_uid)
 
     else:
         return klass.generate_svg(svg_type=svg_type)
 
 
 def gravatar_external(request, gravatar_url_tmpl, email_address, size=30):
     return safe_str(gravatar_url_tmpl)\
         .replace('{email}', email_address) \
         .replace('{md5email}', md5_safe(email_address.lower())) \
         .replace('{netloc}', request.host) \
         .replace('{scheme}', request.scheme) \
         .replace('{size}', safe_str(size))
 
 
 def gravatar_url(email_address, size=30, request=None):
     request = request or get_current_request()
     _use_gravatar = request.call_context.visual.use_gravatar
 
     email_address = email_address or User.DEFAULT_USER_EMAIL
     if isinstance(email_address, unicode):
         # hashlib crashes on unicode items
         email_address = safe_str(email_address)
 
     # empty email or default user
     if not email_address or email_address == User.DEFAULT_USER_EMAIL:
         return initials_gravatar(request, User.DEFAULT_USER_EMAIL, '', '', size=size)
 
     if _use_gravatar:
         gravatar_url_tmpl = request.call_context.visual.gravatar_url \
                             or User.DEFAULT_GRAVATAR_URL
         return gravatar_external(request, gravatar_url_tmpl, email_address, size=size)
 
     else:
         return initials_gravatar(request, email_address, '', '', size=size)
 
 
 def breadcrumb_repo_link(repo):
     """
     Makes a breadcrumbs path link to repo
 
     ex::
         group >> subgroup >> repo
 
     :param repo: a Repository instance
     """
 
     path = [
         link_to(group.name, route_path('repo_group_home', repo_group_name=group.group_name),
                 title='last change:{}'.format(format_date(group.last_commit_change)))
         for group in repo.groups_with_parents
     ] + [
         link_to(repo.just_name, route_path('repo_summary', repo_name=repo.repo_name),
                 title='last change:{}'.format(format_date(repo.last_commit_change)))
     ]
 
     return literal(' &raquo; '.join(path))
 
 
 def breadcrumb_repo_group_link(repo_group):
     """
     Makes a breadcrumbs path link to repo
 
     ex::
         group >> subgroup
 
     :param repo_group: a Repository Group instance
     """
 
     path = [
         link_to(group.name,
                 route_path('repo_group_home', repo_group_name=group.group_name),
                 title='last change:{}'.format(format_date(group.last_commit_change)))
         for group in repo_group.parents
     ] + [
         link_to(repo_group.name,
                 route_path('repo_group_home', repo_group_name=repo_group.group_name),
                 title='last change:{}'.format(format_date(repo_group.last_commit_change)))
     ]
 
     return literal(' &raquo; '.join(path))
 
 
 def format_byte_size_binary(file_size):
     """
     Formats file/folder sizes to standard.
     """
     if file_size is None:
         file_size = 0
 
     formatted_size = format_byte_size(file_size, binary=True)
     return formatted_size
 
 
 def urlify_text(text_, safe=True, **href_attrs):
     """
     Extract urls from text and make html links out of them
     """
 
     url_pat = re.compile(r'''(http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@#.&+]'''
                          '''|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+)''')
 
     def url_func(match_obj):
         url_full = match_obj.groups()[0]
         a_options = dict(href_attrs)
         a_options['href'] = url_full
         a_text = url_full
         return HTML.tag("a", a_text, **a_options)
 
     _new_text = url_pat.sub(url_func, text_)
 
     if safe:
         return literal(_new_text)
     return _new_text
 
 
 def urlify_commits(text_, repo_name):
     """
     Extract commit ids from text and make link from them
 
     :param text_:
     :param repo_name: repo name to build the URL with
     """
 
     url_pat = re.compile(r'(^|\s)([0-9a-fA-F]{12,40})($|\s)')
 
     def url_func(match_obj):
         commit_id = match_obj.groups()[1]
         pref = match_obj.groups()[0]
         suf = match_obj.groups()[2]
 
         tmpl = (
             '%(pref)s<a class="tooltip-hovercard %(cls)s" href="%(url)s" data-hovercard-alt="%(hovercard_alt)s" data-hovercard-url="%(hovercard_url)s">'
             '%(commit_id)s</a>%(suf)s'
         )
         return tmpl % {
             'pref': pref,
             'cls': 'revision-link',
             'url': route_url(
                 'repo_commit', repo_name=repo_name, commit_id=commit_id),
             'commit_id': commit_id,
             'suf': suf,
             'hovercard_alt': 'Commit: {}'.format(commit_id),
             'hovercard_url': route_url(
                 'hovercard_repo_commit', repo_name=repo_name, commit_id=commit_id)
         }
 
     new_text = url_pat.sub(url_func, text_)
 
     return new_text
 
 
 def _process_url_func(match_obj, repo_name, uid, entry,
                       return_raw_data=False, link_format='html'):
     pref = ''
     if match_obj.group().startswith(' '):
         pref = ' '
 
     issue_id = ''.join(match_obj.groups())
 
     if link_format == 'html':
         tmpl = (
             '%(pref)s<a class="tooltip %(cls)s" href="%(url)s" title="%(title)s">'
             '%(issue-prefix)s%(id-repr)s'
             '</a>')
     elif link_format == 'html+hovercard':
         tmpl = (
             '%(pref)s<a class="tooltip-hovercard %(cls)s" href="%(url)s" data-hovercard-url="%(hovercard_url)s">'
             '%(issue-prefix)s%(id-repr)s'
             '</a>')
     elif link_format in ['rst', 'rst+hovercard']:
         tmpl = '`%(issue-prefix)s%(id-repr)s <%(url)s>`_'
     elif link_format in ['markdown', 'markdown+hovercard']:
         tmpl = '[%(pref)s%(issue-prefix)s%(id-repr)s](%(url)s)'
     else:
         raise ValueError('Bad link_format:{}'.format(link_format))
 
     (repo_name_cleaned,
      parent_group_name) = RepoGroupModel()._get_group_name_and_parent(repo_name)
 
     # variables replacement
     named_vars = {
         'id': issue_id,
         'repo': repo_name,
         'repo_name': repo_name_cleaned,
         'group_name': parent_group_name,
         # set dummy keys so we always have them
         'hostname': '',
         'netloc': '',
         'scheme': ''
     }
 
     request = get_current_request()
     if request:
         # exposes, hostname, netloc, scheme
         host_data = get_host_info(request)
         named_vars.update(host_data)
 
     # named regex variables
     named_vars.update(match_obj.groupdict())
     _url = string.Template(entry['url']).safe_substitute(**named_vars)
     desc = string.Template(escape(entry['desc'])).safe_substitute(**named_vars)
     hovercard_url = string.Template(entry.get('hovercard_url', '')).safe_substitute(**named_vars)
 
     def quote_cleaner(input_str):
         """Remove quotes as it's HTML"""
         return input_str.replace('"', '')
 
     data = {
         'pref': pref,
         'cls': quote_cleaner('issue-tracker-link'),
         'url': quote_cleaner(_url),
         'id-repr': issue_id,
         'issue-prefix': entry['pref'],
         'serv': entry['url'],
         'title': bleach.clean(desc, strip=True),
         'hovercard_url': hovercard_url
     }
 
     if return_raw_data:
         return {
             'id': issue_id,
             'url': _url
         }
     return tmpl % data
 
 
 def get_active_pattern_entries(repo_name):
     repo = None
     if repo_name:
         # Retrieving repo_name to avoid invalid repo_name to explode on
         # IssueTrackerSettingsModel but still passing invalid name further down
         repo = Repository.get_by_repo_name(repo_name, cache=True)
 
     settings_model = IssueTrackerSettingsModel(repo=repo)
     active_entries = settings_model.get_settings(cache=True)
     return active_entries
 
 
 pr_pattern_re = regex.compile(r'(?:(?:^!)|(?: !))(\d+)')
 
 allowed_link_formats = [
     'html', 'rst', 'markdown', 'html+hovercard', 'rst+hovercard', 'markdown+hovercard']
 
 compile_cache = {
 
 }
 
 
 def process_patterns(text_string, repo_name, link_format='html', active_entries=None):
 
     if link_format not in allowed_link_formats:
         raise ValueError('Link format can be only one of:{} got {}'.format(
                          allowed_link_formats, link_format))
     issues_data = []
     errors = []
     new_text = text_string
 
     if active_entries is None:
         log.debug('Fetch active issue tracker patterns for repo: %s', repo_name)
         active_entries = get_active_pattern_entries(repo_name)
 
     log.debug('Got %s pattern entries to process', len(active_entries))
 
     for uid, entry in active_entries.items():
 
         if not (entry['pat'] and entry['url']):
             log.debug('skipping due to missing data')
             continue
 
         log.debug('issue tracker entry: uid: `%s` PAT:%s URL:%s PREFIX:%s',
                   uid, entry['pat'], entry['url'], entry['pref'])
 
         if entry.get('pat_compiled'):
             pattern = entry['pat_compiled']
         elif entry['pat'] in compile_cache:
             pattern = compile_cache[entry['pat']]
         else:
             try:
                 pattern = regex.compile(r'%s' % entry['pat'])
             except regex.error as e:
                 regex_err = ValueError('{}:{}'.format(entry['pat'], e))
                 log.exception('issue tracker pattern: `%s` failed to compile', regex_err)
                 errors.append(regex_err)
                 continue
             compile_cache[entry['pat']] = pattern
 
         data_func = partial(
             _process_url_func, repo_name=repo_name, entry=entry, uid=uid,
             return_raw_data=True)
 
         for match_obj in pattern.finditer(text_string):
             issues_data.append(data_func(match_obj))
 
         url_func = partial(
             _process_url_func, repo_name=repo_name, entry=entry, uid=uid,
             link_format=link_format)
 
         new_text = pattern.sub(url_func, new_text)
         log.debug('processed prefix:uid `%s`', uid)
 
     # finally use global replace, eg !123 -> pr-link, those will not catch
     # if already similar pattern exists
     server_url = '${scheme}://${netloc}'
     pr_entry = {
         'pref': '!',
         'url': server_url + '/_admin/pull-requests/${id}',
         'desc': 'Pull Request !${id}',
         'hovercard_url': server_url + '/_hovercard/pull_request/${id}'
     }
     pr_url_func = partial(
         _process_url_func, repo_name=repo_name, entry=pr_entry, uid=None,
         link_format=link_format+'+hovercard')
     new_text = pr_pattern_re.sub(pr_url_func, new_text)
     log.debug('processed !pr pattern')
 
     return new_text, issues_data, errors
 
 
 def urlify_commit_message(commit_text, repository=None, active_pattern_entries=None,
                           issues_container_callback=None, error_container=None):
     """
     Parses given text message and makes proper links.
     issues are linked to given issue-server, and rest is a commit link
     """
 
     def escaper(_text):
         return _text.replace('<', '&lt;').replace('>', '&gt;')
 
     new_text = escaper(commit_text)
 
     # extract http/https links and make them real urls
     new_text = urlify_text(new_text, safe=False)
 
     # urlify commits - extract commit ids and make link out of them, if we have
     # the scope of repository present.
     if repository:
         new_text = urlify_commits(new_text, repository)
 
     # process issue tracker patterns
     new_text, issues, errors = process_patterns(
         new_text, repository or '', active_entries=active_pattern_entries)
 
     if issues_container_callback is not None:
         for issue in issues:
             issues_container_callback(issue)
 
     if error_container is not None:
         error_container.extend(errors)
 
     return literal(new_text)
 
 
 def render_binary(repo_name, file_obj):
     """
     Choose how to render a binary file
     """
 
     # unicode
     filename = file_obj.name
 
     # images
     for ext in ['*.png', '*.jpeg', '*.jpg', '*.ico', '*.gif']:
         if fnmatch.fnmatch(filename, pat=ext):
             src = route_path(
                 'repo_file_raw', repo_name=repo_name,
                 commit_id=file_obj.commit.raw_id,
                 f_path=file_obj.path)
 
             return literal(
                 '<img class="rendered-binary" alt="rendered-image" src="{}">'.format(src))
 
 
 def renderer_from_filename(filename, exclude=None):
     """
     choose a renderer based on filename, this works only for text based files
     """
 
     # ipython
     for ext in ['*.ipynb']:
         if fnmatch.fnmatch(filename, pat=ext):
             return 'jupyter'
 
     is_markup = MarkupRenderer.renderer_from_filename(filename, exclude=exclude)
     if is_markup:
         return is_markup
     return None
 
 
 def render(source, renderer='rst', mentions=False, relative_urls=None,
            repo_name=None, active_pattern_entries=None, issues_container_callback=None):
 
     def maybe_convert_relative_links(html_source):
         if relative_urls:
             return relative_links(html_source, relative_urls)
         return html_source
 
     if renderer == 'plain':
         return literal(
             MarkupRenderer.plain(source, leading_newline=False))
 
     elif renderer == 'rst':
         if repo_name:
             # process patterns on comments if we pass in repo name
             source, issues, errors = process_patterns(
                 source, repo_name, link_format='rst',
                 active_entries=active_pattern_entries)
             if issues_container_callback is not None:
                 for issue in issues:
                     issues_container_callback(issue)
 
         return literal(
             '<div class="rst-block">%s</div>' %
             maybe_convert_relative_links(
                 MarkupRenderer.rst(source, mentions=mentions)))
 
     elif renderer == 'markdown':
         if repo_name:
             # process patterns on comments if we pass in repo name
             source, issues, errors = process_patterns(
                 source, repo_name, link_format='markdown',
                 active_entries=active_pattern_entries)
             if issues_container_callback is not None:
                 for issue in issues:
                     issues_container_callback(issue)
 
 
         return literal(
             '<div class="markdown-block">%s</div>' %
             maybe_convert_relative_links(
                 MarkupRenderer.markdown(source, flavored=True,
                                         mentions=mentions)))
 
     elif renderer == 'jupyter':
         return literal(
             '<div class="ipynb">%s</div>' %
             maybe_convert_relative_links(
                 MarkupRenderer.jupyter(source)))
 
     # None means just show the file-source
     return None
 
 
 def commit_status(repo, commit_id):
     return ChangesetStatusModel().get_status(repo, commit_id)
 
 
 def commit_status_lbl(commit_status):
     return dict(ChangesetStatus.STATUSES).get(commit_status)
 
 
 def commit_time(repo_name, commit_id):
     repo = Repository.get_by_repo_name(repo_name)
     commit = repo.get_commit(commit_id=commit_id)
     return commit.date
 
 
 def get_permission_name(key):
     return dict(Permission.PERMS).get(key)
 
 
 def journal_filter_help(request):
     _ = request.translate
     from rhodecode.lib.audit_logger import ACTIONS
     actions = '\n'.join(textwrap.wrap(', '.join(sorted(ACTIONS.keys())), 80))
 
     return _(
         'Example filter terms:\n' +
         '     repository:vcs\n' +
         '     username:marcin\n' +
         '     username:(NOT marcin)\n' +
         '     action:*push*\n' +
         '     ip:127.0.0.1\n' +
         '     date:20120101\n' +
         '     date:[20120101100000 TO 20120102]\n' +
         '\n' +
         'Actions: {actions}\n' +
         '\n' +
         'Generate wildcards using \'*\' character:\n' +
         '     "repository:vcs*" - search everything starting with \'vcs\'\n' +
         '     "repository:*vcs*" - search for repository containing \'vcs\'\n' +
         '\n' +
         'Optional AND / OR operators in queries\n' +
         '     "repository:vcs OR repository:test"\n' +
         '     "username:test AND repository:test*"\n'
     ).format(actions=actions)
 
 
 def not_mapped_error(repo_name):
     from rhodecode.translation import _
     flash(_('%s repository is not mapped to db perhaps'
             ' it was created or renamed from the filesystem'
             ' please run the application again'
             ' in order to rescan repositories') % repo_name, category='error')
 
 
 def ip_range(ip_addr):
     from rhodecode.model.db import UserIpMap
     s, e = UserIpMap._get_ip_range(ip_addr)
     return '%s - %s' % (s, e)
 
 
 def form(url, method='post', needs_csrf_token=True, **attrs):
     """Wrapper around webhelpers.tags.form to prevent CSRF attacks."""
     if method.lower() != 'get' and needs_csrf_token:
         raise Exception(
             'Forms to POST/PUT/DELETE endpoints should have (in general) a ' +
             'CSRF token. If the endpoint does not require such token you can ' +
             'explicitly set the parameter needs_csrf_token to false.')
 
     return insecure_form(url, method=method, **attrs)
 
 
 def secure_form(form_url, method="POST", multipart=False, **attrs):
     """Start a form tag that points the action to an url. This
     form tag will also include the hidden field containing
     the auth token.
 
     The url options should be given either as a string, or as a
     ``url()`` function. The method for the form defaults to POST.
 
     Options:
 
     ``multipart``
         If set to True, the enctype is set to "multipart/form-data".
     ``method``
         The method to use when submitting the form, usually either
         "GET" or "POST". If "PUT", "DELETE", or another verb is used, a
         hidden input with name _method is added to simulate the verb
         over POST.
 
     """
 
     if 'request' in attrs:
         session = attrs['request'].session
         del attrs['request']
     else:
         raise ValueError(
             'Calling this form requires request= to be passed as argument')
 
     _form = insecure_form(form_url, method, multipart, **attrs)
     token = literal(
         '<input type="hidden" name="{}" value="{}">'.format(
             csrf_token_key, get_csrf_token(session)))
 
     return literal("%s\n%s" % (_form, token))
 
 
 def dropdownmenu(name, selected, options, enable_filter=False, **attrs):
     select_html = select(name, selected, options, **attrs)
 
     select2 = """
         <script>
             $(document).ready(function() {
                   $('#%s').select2({
                       containerCssClass: 'drop-menu %s',
                       dropdownCssClass: 'drop-menu-dropdown',
                       dropdownAutoWidth: true%s
                   });
             });
         </script>
     """
 
     filter_option = """,
                         minimumResultsForSearch: -1
     """
     input_id = attrs.get('id') or name
     extra_classes = ' '.join(attrs.pop('extra_classes', []))
     filter_enabled = "" if enable_filter else filter_option
     select_script = literal(select2 % (input_id, extra_classes, filter_enabled))
 
     return literal(select_html+select_script)
 
 
 def get_visual_attr(tmpl_context_var, attr_name):
     """
     A safe way to get a variable from visual variable of template context
 
     :param tmpl_context_var: instance of tmpl_context, usually present as `c`
     :param attr_name: name of the attribute we fetch from the c.visual
     """
     visual = getattr(tmpl_context_var, 'visual', None)
     if not visual:
         return
     else:
         return getattr(visual, attr_name, None)
 
 
 def get_last_path_part(file_node):
     if not file_node.path:
         return u'/'
 
     path = safe_unicode(file_node.path.split('/')[-1])
     return u'../' + path
 
 
 def route_url(*args, **kwargs):
     """
     Wrapper around pyramids `route_url` (fully qualified url) function.
     """
     req = get_current_request()
     return req.route_url(*args, **kwargs)
 
 
 def route_path(*args, **kwargs):
     """
     Wrapper around pyramids `route_path` function.
     """
     req = get_current_request()
     return req.route_path(*args, **kwargs)
 
 
 def route_path_or_none(*args, **kwargs):
     try:
         return route_path(*args, **kwargs)
     except KeyError:
         return None
 
 
 def current_route_path(request, **kw):
     new_args = request.GET.mixed()
     new_args.update(kw)
     return request.current_route_path(_query=new_args)
 
 
 def curl_api_example(method, args):
     args_json = json.dumps(OrderedDict([
         ('id', 1),
         ('auth_token', 'SECRET'),
         ('method', method),
         ('args', args)
     ]))
 
     return "curl {api_url} -X POST -H 'content-type:text/plain' --data-binary '{args_json}'".format(
         api_url=route_url('apiv2'),
         args_json=args_json
     )
 
 
 def api_call_example(method, args):
     """
     Generates an API call example via CURL
     """
     curl_call = curl_api_example(method, args)
 
     return literal(
         curl_call +
         "<br/><br/>SECRET can be found in <a href=\"{token_url}\">auth-tokens</a> page, "
         "and needs to be of `api calls` role."
         .format(token_url=route_url('my_account_auth_tokens')))
 
 
 def notification_description(notification, request):
     """
     Generate notification human readable description based on notification type
     """
     from rhodecode.model.notification import NotificationModel
     return NotificationModel().make_description(
         notification, translate=request.translate)
 
 
 def go_import_header(request, db_repo=None):
     """
     Creates a header for go-import functionality in Go Lang
     """
 
     if not db_repo:
         return
     if 'go-get' not in request.GET:
         return
 
     clone_url = db_repo.clone_url()
     prefix = re.split(r'^https?:\/\/', clone_url)[-1]
     # we have a repo and go-get flag,
     return literal('<meta name="go-import" content="{} {} {}">'.format(
         prefix, db_repo.repo_type, clone_url))
 
 
 def reviewer_as_json(*args, **kwargs):
     from rhodecode.apps.repository.utils import reviewer_as_json as _reviewer_as_json
     return _reviewer_as_json(*args, **kwargs)
 
 
 def get_repo_view_type(request):
     route_name = request.matched_route.name
     route_to_view_type = {
         'repo_changelog': 'commits',
         'repo_commits': 'commits',
         'repo_files': 'files',
         'repo_summary': 'summary',
         'repo_commit': 'commit'
     }
 
     return route_to_view_type.get(route_name)
 
 
 def is_active(menu_entry, selected):
     """
     Returns active class for selecting menus in templates
     <li class=${h.is_active('settings', current_active)}></li>
     """
     if not isinstance(menu_entry, list):
         menu_entry = [menu_entry]
 
     if selected in menu_entry:
         return "active"
 
 
 class IssuesRegistry(object):
     """
     issue_registry = IssuesRegistry()
     some_func(issues_callback=issues_registry(...))
     """
 
     def __init__(self):
         self.issues = []
         self.unique_issues = collections.defaultdict(lambda: [])
 
     def __call__(self, commit_dict=None):
         def callback(issue):
             if commit_dict and issue:
                 issue['commit'] = commit_dict
             self.issues.append(issue)
             self.unique_issues[issue['id']].append(issue)
         return callback
 
     def get_issues(self):
         return self.issues
 
     @property
     def issues_unique_count(self):
         return len(set(i['id'] for i in self.issues))

rhodecode/lib/middleware/simplevcs.py

@@ -1,679 +1,679 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2014-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 SimpleVCS middleware for handling protocol request (push/clone etc.)
 It's implemented with basic auth function
 """
 
 import os
 import re
 import logging
 import importlib
 from functools import wraps
-from StringIO import StringIO
+from io import StringIO
 from lxml import etree
 
 import time
 from paste.httpheaders import REMOTE_USER, AUTH_TYPE
 
 from pyramid.httpexceptions import (
     HTTPNotFound, HTTPForbidden, HTTPNotAcceptable, HTTPInternalServerError)
 from zope.cachedescriptors.property import Lazy as LazyProperty
 
 import rhodecode
 from rhodecode.authentication.base import authenticate, VCS_TYPE, loadplugin
 from rhodecode.lib import rc_cache
 from rhodecode.lib.auth import AuthUser, HasPermissionAnyMiddleware
 from rhodecode.lib.base import (
     BasicAuth, get_ip_addr, get_user_agent, vcs_operation_context)
 from rhodecode.lib.exceptions import (UserCreationError, NotAllowedToCreateUserError)
 from rhodecode.lib.hooks_daemon import prepare_callback_daemon
 from rhodecode.lib.middleware import appenlight
 from rhodecode.lib.middleware.utils import scm_app_http
 from rhodecode.lib.utils import is_valid_repo, SLUG_RE
 from rhodecode.lib.utils2 import safe_str, fix_PATH, str2bool, safe_unicode
 from rhodecode.lib.vcs.conf import settings as vcs_settings
 from rhodecode.lib.vcs.backends import base
 
 from rhodecode.model import meta
 from rhodecode.model.db import User, Repository, PullRequest
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.pull_request import PullRequestModel
 from rhodecode.model.settings import SettingsModel, VcsSettingsModel
 
 log = logging.getLogger(__name__)
 
 
 def extract_svn_txn_id(acl_repo_name, data):
     """
     Helper method for extraction of svn txn_id from submitted XML data during
     POST operations
     """
     try:
         root = etree.fromstring(data)
         pat = re.compile(r'/txn/(?P<txn_id>.*)')
         for el in root:
             if el.tag == '{DAV:}source':
                 for sub_el in el:
                     if sub_el.tag == '{DAV:}href':
                         match = pat.search(sub_el.text)
                         if match:
                             svn_tx_id = match.groupdict()['txn_id']
                             txn_id = rc_cache.utils.compute_key_from_params(
                                 acl_repo_name, svn_tx_id)
                             return txn_id
     except Exception:
         log.exception('Failed to extract txn_id')
 
 
 def initialize_generator(factory):
     """
     Initializes the returned generator by draining its first element.
 
     This can be used to give a generator an initializer, which is the code
     up to the first yield statement. This decorator enforces that the first
     produced element has the value ``"__init__"`` to make its special
     purpose very explicit in the using code.
     """
 
     @wraps(factory)
     def wrapper(*args, **kwargs):
         gen = factory(*args, **kwargs)
         try:
             init = gen.next()
         except StopIteration:
             raise ValueError('Generator must yield at least one element.')
         if init != "__init__":
             raise ValueError('First yielded element must be "__init__".')
         return gen
     return wrapper
 
 
 class SimpleVCS(object):
     """Common functionality for SCM HTTP handlers."""
 
     SCM = 'unknown'
 
     acl_repo_name = None
     url_repo_name = None
     vcs_repo_name = None
     rc_extras = {}
 
     # We have to handle requests to shadow repositories different than requests
     # to normal repositories. Therefore we have to distinguish them. To do this
     # we use this regex which will match only on URLs pointing to shadow
     # repositories.
     shadow_repo_re = re.compile(
         '(?P<groups>(?:{slug_pat}/)*)'  # repo groups
         '(?P<target>{slug_pat})/'       # target repo
         'pull-request/(?P<pr_id>\d+)/'  # pull request
         'repository$'                   # shadow repo
         .format(slug_pat=SLUG_RE.pattern))
 
     def __init__(self, config, registry):
         self.registry = registry
         self.config = config
         # re-populated by specialized middleware
         self.repo_vcs_config = base.Config()
 
         rc_settings = SettingsModel().get_all_settings(cache=True, from_request=False)
         realm = rc_settings.get('rhodecode_realm') or 'RhodeCode AUTH'
 
         # authenticate this VCS request using authfunc
         auth_ret_code_detection = \
             str2bool(self.config.get('auth_ret_code_detection', False))
         self.authenticate = BasicAuth(
             '', authenticate, registry, config.get('auth_ret_code'),
             auth_ret_code_detection, rc_realm=realm)
         self.ip_addr = '0.0.0.0'
 
     @LazyProperty
     def global_vcs_config(self):
         try:
             return VcsSettingsModel().get_ui_settings_as_config_obj()
         except Exception:
             return base.Config()
 
     @property
     def base_path(self):
         settings_path = self.repo_vcs_config.get(*VcsSettingsModel.PATH_SETTING)
 
         if not settings_path:
             settings_path = self.global_vcs_config.get(*VcsSettingsModel.PATH_SETTING)
 
         if not settings_path:
             # try, maybe we passed in explicitly as config option
             settings_path = self.config.get('base_path')
 
         if not settings_path:
             raise ValueError('FATAL: base_path is empty')
         return settings_path
 
     def set_repo_names(self, environ):
         """
         This will populate the attributes acl_repo_name, url_repo_name,
         vcs_repo_name and is_shadow_repo. In case of requests to normal (non
         shadow) repositories all names are equal. In case of requests to a
         shadow repository the acl-name points to the target repo of the pull
         request and the vcs-name points to the shadow repo file system path.
         The url-name is always the URL used by the vcs client program.
 
         Example in case of a shadow repo:
             acl_repo_name = RepoGroup/MyRepo
             url_repo_name = RepoGroup/MyRepo/pull-request/3/repository
             vcs_repo_name = /repo/base/path/RepoGroup/.__shadow_MyRepo_pr-3'
         """
         # First we set the repo name from URL for all attributes. This is the
         # default if handling normal (non shadow) repo requests.
         self.url_repo_name = self._get_repository_name(environ)
         self.acl_repo_name = self.vcs_repo_name = self.url_repo_name
         self.is_shadow_repo = False
 
         # Check if this is a request to a shadow repository.
         match = self.shadow_repo_re.match(self.url_repo_name)
         if match:
             match_dict = match.groupdict()
 
             # Build acl repo name from regex match.
             acl_repo_name = safe_unicode('{groups}{target}'.format(
                 groups=match_dict['groups'] or '',
                 target=match_dict['target']))
 
             # Retrieve pull request instance by ID from regex match.
             pull_request = PullRequest.get(match_dict['pr_id'])
 
             # Only proceed if we got a pull request and if acl repo name from
             # URL equals the target repo name of the pull request.
             if pull_request and (acl_repo_name == pull_request.target_repo.repo_name):
 
                 # Get file system path to shadow repository.
                 workspace_id = PullRequestModel()._workspace_id(pull_request)
                 vcs_repo_name = pull_request.target_repo.get_shadow_repository_path(workspace_id)
 
                 # Store names for later usage.
                 self.vcs_repo_name = vcs_repo_name
                 self.acl_repo_name = acl_repo_name
                 self.is_shadow_repo = True
 
         log.debug('Setting all VCS repository names: %s', {
             'acl_repo_name': self.acl_repo_name,
             'url_repo_name': self.url_repo_name,
             'vcs_repo_name': self.vcs_repo_name,
         })
 
     @property
     def scm_app(self):
         custom_implementation = self.config['vcs.scm_app_implementation']
         if custom_implementation == 'http':
             log.debug('Using HTTP implementation of scm app.')
             scm_app_impl = scm_app_http
         else:
             log.debug('Using custom implementation of scm_app: "{}"'.format(
                 custom_implementation))
             scm_app_impl = importlib.import_module(custom_implementation)
         return scm_app_impl
 
     def _get_by_id(self, repo_name):
         """
         Gets a special pattern _<ID> from clone url and tries to replace it
         with a repository_name for support of _<ID> non changeable urls
         """
 
         data = repo_name.split('/')
         if len(data) >= 2:
             from rhodecode.model.repo import RepoModel
             by_id_match = RepoModel().get_repo_by_id(repo_name)
             if by_id_match:
                 data[1] = by_id_match.repo_name
 
         return safe_str('/'.join(data))
 
     def _invalidate_cache(self, repo_name):
         """
         Set's cache for this repository for invalidation on next access
 
         :param repo_name: full repo name, also a cache key
         """
         ScmModel().mark_for_invalidation(repo_name)
 
     def is_valid_and_existing_repo(self, repo_name, base_path, scm_type):
         db_repo = Repository.get_by_repo_name(repo_name)
         if not db_repo:
             log.debug('Repository `%s` not found inside the database.',
                       repo_name)
             return False
 
         if db_repo.repo_type != scm_type:
             log.warning(
                 'Repository `%s` have incorrect scm_type, expected %s got %s',
                 repo_name, db_repo.repo_type, scm_type)
             return False
 
         config = db_repo._config
         config.set('extensions', 'largefiles', '')
         return is_valid_repo(
             repo_name, base_path,
             explicit_scm=scm_type, expect_scm=scm_type, config=config)
 
     def valid_and_active_user(self, user):
         """
         Checks if that user is not empty, and if it's actually object it checks
         if he's active.
 
         :param user: user object or None
         :return: boolean
         """
         if user is None:
             return False
 
         elif user.active:
             return True
 
         return False
 
     @property
     def is_shadow_repo_dir(self):
         return os.path.isdir(self.vcs_repo_name)
 
     def _check_permission(self, action, user, auth_user, repo_name, ip_addr=None,
                           plugin_id='', plugin_cache_active=False, cache_ttl=0):
         """
         Checks permissions using action (push/pull) user and repository
         name. If plugin_cache and ttl is set it will use the plugin which
         authenticated the user to store the cached permissions result for N
         amount of seconds as in cache_ttl
 
         :param action: push or pull action
         :param user: user instance
         :param repo_name: repository name
         """
 
         log.debug('AUTH_CACHE_TTL for permissions `%s` active: %s (TTL: %s)',
                   plugin_id, plugin_cache_active, cache_ttl)
 
         user_id = user.user_id
         cache_namespace_uid = 'cache_user_auth.{}'.format(user_id)
         region = rc_cache.get_or_create_region('cache_perms', cache_namespace_uid)
 
         @region.conditional_cache_on_arguments(namespace=cache_namespace_uid,
                                                expiration_time=cache_ttl,
                                                condition=plugin_cache_active)
         def compute_perm_vcs(
                 cache_name, plugin_id, action, user_id, repo_name, ip_addr):
 
             log.debug('auth: calculating permission access now...')
             # check IP
             inherit = user.inherit_default_permissions
             ip_allowed = AuthUser.check_ip_allowed(
                 user_id, ip_addr, inherit_from_default=inherit)
             if ip_allowed:
                 log.info('Access for IP:%s allowed', ip_addr)
             else:
                 return False
 
             if action == 'push':
                 perms = ('repository.write', 'repository.admin')
                 if not HasPermissionAnyMiddleware(*perms)(auth_user, repo_name):
                     return False
 
             else:
                 # any other action need at least read permission
                 perms = (
                     'repository.read', 'repository.write', 'repository.admin')
                 if not HasPermissionAnyMiddleware(*perms)(auth_user, repo_name):
                     return False
 
             return True
 
         start = time.time()
         log.debug('Running plugin `%s` permissions check', plugin_id)
 
         # for environ based auth, password can be empty, but then the validation is
         # on the server that fills in the env data needed for authentication
         perm_result = compute_perm_vcs(
             'vcs_permissions', plugin_id, action, user.user_id, repo_name, ip_addr)
 
         auth_time = time.time() - start
         log.debug('Permissions for plugin `%s` completed in %.4fs, '
                   'expiration time of fetched cache %.1fs.',
                   plugin_id, auth_time, cache_ttl)
 
         return perm_result
 
     def _get_http_scheme(self, environ):
         try:
             return environ['wsgi.url_scheme']
         except Exception:
             log.exception('Failed to read http scheme')
             return 'http'
 
     def _check_ssl(self, environ, start_response):
         """
         Checks the SSL check flag and returns False if SSL is not present
         and required True otherwise
         """
         org_proto = environ['wsgi._org_proto']
         # check if we have SSL required  ! if not it's a bad request !
         require_ssl = str2bool(self.repo_vcs_config.get('web', 'push_ssl'))
         if require_ssl and org_proto == 'http':
             log.debug(
                 'Bad request: detected protocol is `%s` and '
                 'SSL/HTTPS is required.', org_proto)
             return False
         return True
 
     def _get_default_cache_ttl(self):
         # take AUTH_CACHE_TTL from the `rhodecode` auth plugin
         plugin = loadplugin('egg:rhodecode-enterprise-ce#rhodecode')
         plugin_settings = plugin.get_settings()
         plugin_cache_active, cache_ttl = plugin.get_ttl_cache(
             plugin_settings) or (False, 0)
         return plugin_cache_active, cache_ttl
 
     def __call__(self, environ, start_response):
         try:
             return self._handle_request(environ, start_response)
         except Exception:
             log.exception("Exception while handling request")
             appenlight.track_exception(environ)
             return HTTPInternalServerError()(environ, start_response)
         finally:
             meta.Session.remove()
 
     def _handle_request(self, environ, start_response):
         if not self._check_ssl(environ, start_response):
             reason = ('SSL required, while RhodeCode was unable '
                       'to detect this as SSL request')
             log.debug('User not allowed to proceed, %s', reason)
             return HTTPNotAcceptable(reason)(environ, start_response)
 
         if not self.url_repo_name:
             log.warning('Repository name is empty: %s', self.url_repo_name)
             # failed to get repo name, we fail now
             return HTTPNotFound()(environ, start_response)
         log.debug('Extracted repo name is %s', self.url_repo_name)
 
         ip_addr = get_ip_addr(environ)
         user_agent = get_user_agent(environ)
         username = None
 
         # skip passing error to error controller
         environ['pylons.status_code_redirect'] = True
 
         # ======================================================================
         # GET ACTION PULL or PUSH
         # ======================================================================
         action = self._get_action(environ)
 
         # ======================================================================
         # Check if this is a request to a shadow repository of a pull request.
         # In this case only pull action is allowed.
         # ======================================================================
         if self.is_shadow_repo and action != 'pull':
             reason = 'Only pull action is allowed for shadow repositories.'
             log.debug('User not allowed to proceed, %s', reason)
             return HTTPNotAcceptable(reason)(environ, start_response)
 
         # Check if the shadow repo actually exists, in case someone refers
         # to it, and it has been deleted because of successful merge.
         if self.is_shadow_repo and not self.is_shadow_repo_dir:
             log.debug(
                 'Shadow repo detected, and shadow repo dir `%s` is missing',
                 self.is_shadow_repo_dir)
             return HTTPNotFound()(environ, start_response)
 
         # ======================================================================
         # CHECK ANONYMOUS PERMISSION
         # ======================================================================
         detect_force_push = False
         check_branch_perms = False
         if action in ['pull', 'push']:
             user_obj = anonymous_user = User.get_default_user()
             auth_user = user_obj.AuthUser()
             username = anonymous_user.username
             if anonymous_user.active:
                 plugin_cache_active, cache_ttl = self._get_default_cache_ttl()
                 # ONLY check permissions if the user is activated
                 anonymous_perm = self._check_permission(
                     action, anonymous_user, auth_user, self.acl_repo_name, ip_addr,
                     plugin_id='anonymous_access',
                     plugin_cache_active=plugin_cache_active,
                     cache_ttl=cache_ttl,
                 )
             else:
                 anonymous_perm = False
 
             if not anonymous_user.active or not anonymous_perm:
                 if not anonymous_user.active:
                     log.debug('Anonymous access is disabled, running '
                               'authentication')
 
                 if not anonymous_perm:
                     log.debug('Not enough credentials to access this '
                               'repository as anonymous user')
 
                 username = None
                 # ==============================================================
                 # DEFAULT PERM FAILED OR ANONYMOUS ACCESS IS DISABLED SO WE
                 # NEED TO AUTHENTICATE AND ASK FOR AUTH USER PERMISSIONS
                 # ==============================================================
 
                 # try to auth based on environ, container auth methods
                 log.debug('Running PRE-AUTH for container based authentication')
                 pre_auth = authenticate(
                     '', '', environ, VCS_TYPE, registry=self.registry,
                     acl_repo_name=self.acl_repo_name)
                 if pre_auth and pre_auth.get('username'):
                     username = pre_auth['username']
                 log.debug('PRE-AUTH got %s as username', username)
                 if pre_auth:
                     log.debug('PRE-AUTH successful from %s',
                               pre_auth.get('auth_data', {}).get('_plugin'))
 
                 # If not authenticated by the container, running basic auth
                 # before inject the calling repo_name for special scope checks
                 self.authenticate.acl_repo_name = self.acl_repo_name
 
                 plugin_cache_active, cache_ttl = False, 0
                 plugin = None
                 if not username:
                     self.authenticate.realm = self.authenticate.get_rc_realm()
 
                     try:
                         auth_result = self.authenticate(environ)
                     except (UserCreationError, NotAllowedToCreateUserError) as e:
                         log.error(e)
                         reason = safe_str(e)
                         return HTTPNotAcceptable(reason)(environ, start_response)
 
                     if isinstance(auth_result, dict):
                         AUTH_TYPE.update(environ, 'basic')
                         REMOTE_USER.update(environ, auth_result['username'])
                         username = auth_result['username']
                         plugin = auth_result.get('auth_data', {}).get('_plugin')
                         log.info(
                             'MAIN-AUTH successful for user `%s` from %s plugin',
                             username, plugin)
 
                         plugin_cache_active, cache_ttl = auth_result.get(
                             'auth_data', {}).get('_ttl_cache') or (False, 0)
                     else:
                         return auth_result.wsgi_application(environ, start_response)
 
                 # ==============================================================
                 # CHECK PERMISSIONS FOR THIS REQUEST USING GIVEN USERNAME
                 # ==============================================================
                 user = User.get_by_username(username)
                 if not self.valid_and_active_user(user):
                     return HTTPForbidden()(environ, start_response)
                 username = user.username
                 user_id = user.user_id
 
                 # check user attributes for password change flag
                 user_obj = user
                 auth_user = user_obj.AuthUser()
                 if user_obj and user_obj.username != User.DEFAULT_USER and \
                         user_obj.user_data.get('force_password_change'):
                     reason = 'password change required'
                     log.debug('User not allowed to authenticate, %s', reason)
                     return HTTPNotAcceptable(reason)(environ, start_response)
 
                 # check permissions for this repository
                 perm = self._check_permission(
                     action, user, auth_user, self.acl_repo_name, ip_addr,
                     plugin, plugin_cache_active, cache_ttl)
                 if not perm:
                     return HTTPForbidden()(environ, start_response)
                 environ['rc_auth_user_id'] = user_id
 
             if action == 'push':
                 perms = auth_user.get_branch_permissions(self.acl_repo_name)
                 if perms:
                     check_branch_perms = True
                     detect_force_push = True
 
         # extras are injected into UI object and later available
         # in hooks executed by RhodeCode
         check_locking = _should_check_locking(environ.get('QUERY_STRING'))
 
         extras = vcs_operation_context(
             environ, repo_name=self.acl_repo_name, username=username,
             action=action, scm=self.SCM, check_locking=check_locking,
             is_shadow_repo=self.is_shadow_repo, check_branch_perms=check_branch_perms,
             detect_force_push=detect_force_push
         )
 
         # ======================================================================
         # REQUEST HANDLING
         # ======================================================================
         repo_path = os.path.join(
             safe_str(self.base_path), safe_str(self.vcs_repo_name))
         log.debug('Repository path is %s', repo_path)
 
         fix_PATH()
 
         log.info(
             '%s action on %s repo "%s" by "%s" from %s %s',
             action, self.SCM, safe_str(self.url_repo_name),
             safe_str(username), ip_addr, user_agent)
 
         return self._generate_vcs_response(
             environ, start_response, repo_path, extras, action)
 
     @initialize_generator
     def _generate_vcs_response(
             self, environ, start_response, repo_path, extras, action):
         """
         Returns a generator for the response content.
 
         This method is implemented as a generator, so that it can trigger
         the cache validation after all content sent back to the client. It
         also handles the locking exceptions which will be triggered when
         the first chunk is produced by the underlying WSGI application.
         """
         txn_id = ''
         if 'CONTENT_LENGTH' in environ and environ['REQUEST_METHOD'] == 'MERGE':
             # case for SVN, we want to re-use the callback daemon port
             # so we use the txn_id, for this we peek the body, and still save
             # it as wsgi.input
             data = environ['wsgi.input'].read()
             environ['wsgi.input'] = StringIO(data)
             txn_id = extract_svn_txn_id(self.acl_repo_name, data)
 
         callback_daemon, extras = self._prepare_callback_daemon(
             extras, environ, action, txn_id=txn_id)
         log.debug('HOOKS extras is %s', extras)
 
         http_scheme = self._get_http_scheme(environ)
 
         config = self._create_config(extras, self.acl_repo_name, scheme=http_scheme)
         app = self._create_wsgi_app(repo_path, self.url_repo_name, config)
         with callback_daemon:
             app.rc_extras = extras
 
             try:
                 response = app(environ, start_response)
             finally:
                 # This statement works together with the decorator
                 # "initialize_generator" above. The decorator ensures that
                 # we hit the first yield statement before the generator is
                 # returned back to the WSGI server. This is needed to
                 # ensure that the call to "app" above triggers the
                 # needed callback to "start_response" before the
                 # generator is actually used.
                 yield "__init__"
 
             # iter content
             for chunk in response:
                 yield chunk
 
             try:
                 # invalidate cache on push
                 if action == 'push':
                     self._invalidate_cache(self.url_repo_name)
             finally:
                 meta.Session.remove()
 
     def _get_repository_name(self, environ):
         """Get repository name out of the environmnent
 
         :param environ: WSGI environment
         """
         raise NotImplementedError()
 
     def _get_action(self, environ):
         """Map request commands into a pull or push command.
 
         :param environ: WSGI environment
         """
         raise NotImplementedError()
 
     def _create_wsgi_app(self, repo_path, repo_name, config):
         """Return the WSGI app that will finally handle the request."""
         raise NotImplementedError()
 
     def _create_config(self, extras, repo_name, scheme='http'):
         """Create a safe config representation."""
         raise NotImplementedError()
 
     def _should_use_callback_daemon(self, extras, environ, action):
         if extras.get('is_shadow_repo'):
             # we don't want to execute hooks, and callback daemon for shadow repos
             return False
         return True
 
     def _prepare_callback_daemon(self, extras, environ, action, txn_id=None):
         direct_calls = vcs_settings.HOOKS_DIRECT_CALLS
         if not self._should_use_callback_daemon(extras, environ, action):
             # disable callback daemon for actions that don't require it
             direct_calls = True
 
         return prepare_callback_daemon(
             extras, protocol=vcs_settings.HOOKS_PROTOCOL,
             host=vcs_settings.HOOKS_HOST, use_direct_calls=direct_calls, txn_id=txn_id)
 
 
 def _should_check_locking(query_string):
     # this is kind of hacky, but due to how mercurial handles client-server
     # server see all operation on commit; bookmarks, phases and
     # obsolescence marker in different transaction, we don't want to check
     # locking on those
     return query_string not in ['cmd=listkeys']

rhodecode/lib/vcs/backends/git/commit.py

@@ -1,496 +1,496 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2014-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 GIT commit module
 """
 
 import re
 import stat
 from itertools import chain
-from StringIO import StringIO
+from io import StringIO
 
 from zope.cachedescriptors.property import Lazy as LazyProperty
 
 from rhodecode.lib.datelib import utcdate_fromtimestamp
 from rhodecode.lib.utils import safe_unicode, safe_str
 from rhodecode.lib.utils2 import safe_int
 from rhodecode.lib.vcs.conf import settings
 from rhodecode.lib.vcs.backends import base
 from rhodecode.lib.vcs.exceptions import CommitError, NodeDoesNotExistError
 from rhodecode.lib.vcs.nodes import (
     FileNode, DirNode, NodeKind, RootNode, SubModuleNode,
     ChangedFileNodesGenerator, AddedFileNodesGenerator,
     RemovedFileNodesGenerator, LargeFileNode)
 from rhodecode.lib.vcs.compat import configparser
 
 
 class GitCommit(base.BaseCommit):
     """
     Represents state of the repository at single commit id.
     """
 
     _filter_pre_load = [
         # done through a more complex tree walk on parents
         "affected_files",
         # done through subprocess not remote call
         "children",
         # done through a more complex tree walk on parents
         "status",
         # mercurial specific property not supported here
         "_file_paths",
         # mercurial specific property not supported here
         'obsolete',
         # mercurial specific property not supported here
         'phase',
         # mercurial specific property not supported here
         'hidden'
     ]
 
     def __init__(self, repository, raw_id, idx, pre_load=None):
         self.repository = repository
         self._remote = repository._remote
         # TODO: johbo: Tweak of raw_id should not be necessary
         self.raw_id = safe_str(raw_id)
         self.idx = idx
 
         self._set_bulk_properties(pre_load)
 
         # caches
         self._stat_modes = {}  # stat info for paths
         self._paths = {}  # path processed with parse_tree
         self.nodes = {}
         self._submodules = None
 
     def _set_bulk_properties(self, pre_load):
 
         if not pre_load:
             return
         pre_load = [entry for entry in pre_load
                     if entry not in self._filter_pre_load]
         if not pre_load:
             return
 
         result = self._remote.bulk_request(self.raw_id, pre_load)
         for attr, value in result.items():
             if attr in ["author", "message"]:
                 if value:
                     value = safe_unicode(value)
             elif attr == "date":
                 value = utcdate_fromtimestamp(*value)
             elif attr == "parents":
                 value = self._make_commits(value)
             elif attr == "branch":
                 value = self._set_branch(value)
             self.__dict__[attr] = value
 
     @LazyProperty
     def _commit(self):
         return self._remote[self.raw_id]
 
     @LazyProperty
     def _tree_id(self):
         return self._remote[self._commit['tree']]['id']
 
     @LazyProperty
     def id(self):
         return self.raw_id
 
     @LazyProperty
     def short_id(self):
         return self.raw_id[:12]
 
     @LazyProperty
     def message(self):
         return safe_unicode(self._remote.message(self.id))
 
     @LazyProperty
     def committer(self):
         return safe_unicode(self._remote.author(self.id))
 
     @LazyProperty
     def author(self):
         return safe_unicode(self._remote.author(self.id))
 
     @LazyProperty
     def date(self):
         unix_ts, tz = self._remote.date(self.raw_id)
         return utcdate_fromtimestamp(unix_ts, tz)
 
     @LazyProperty
     def status(self):
         """
         Returns modified, added, removed, deleted files for current commit
         """
         return self.changed, self.added, self.removed
 
     @LazyProperty
     def tags(self):
         tags = [safe_unicode(name) for name,
                 commit_id in self.repository.tags.iteritems()
                 if commit_id == self.raw_id]
         return tags
 
     @LazyProperty
     def commit_branches(self):
         branches = []
         for name, commit_id in self.repository.branches.iteritems():
             if commit_id == self.raw_id:
                 branches.append(name)
         return branches
 
     def _set_branch(self, branches):
         if branches:
             # actually commit can have multiple branches in git
             return safe_unicode(branches[0])
 
     @LazyProperty
     def branch(self):
         branches = self._remote.branch(self.raw_id)
         return self._set_branch(branches)
 
     def _get_tree_id_for_path(self, path):
         path = safe_str(path)
         if path in self._paths:
             return self._paths[path]
 
         tree_id = self._tree_id
 
         path = path.strip('/')
         if path == '':
             data = [tree_id, "tree"]
             self._paths[''] = data
             return data
 
         tree_id, tree_type, tree_mode = \
             self._remote.tree_and_type_for_path(self.raw_id, path)
         if tree_id is None:
             raise self.no_node_at_path(path)
 
         self._paths[path] = [tree_id, tree_type]
         self._stat_modes[path] = tree_mode
 
         if path not in self._paths:
             raise self.no_node_at_path(path)
 
         return self._paths[path]
 
     def _get_kind(self, path):
         tree_id, type_ = self._get_tree_id_for_path(path)
         if type_ == 'blob':
             return NodeKind.FILE
         elif type_ == 'tree':
             return NodeKind.DIR
         elif type_ == 'link':
             return NodeKind.SUBMODULE
         return None
 
     def _get_filectx(self, path):
         path = self._fix_path(path)
         if self._get_kind(path) != NodeKind.FILE:
             raise CommitError(
                 "File does not exist for commit %s at  '%s'" % (self.raw_id, path))
         return path
 
     def _get_file_nodes(self):
         return chain(*(t[2] for t in self.walk()))
 
     @LazyProperty
     def parents(self):
         """
         Returns list of parent commits.
         """
         parent_ids = self._remote.parents(self.id)
         return self._make_commits(parent_ids)
 
     @LazyProperty
     def children(self):
         """
         Returns list of child commits.
         """
 
         children = self._remote.children(self.raw_id)
         return self._make_commits(children)
 
     def _make_commits(self, commit_ids):
         def commit_maker(_commit_id):
             return self.repository.get_commit(commit_id=commit_id)
 
         return [commit_maker(commit_id) for commit_id in commit_ids]
 
     def get_file_mode(self, path):
         """
         Returns stat mode of the file at the given `path`.
         """
         path = safe_str(path)
         # ensure path is traversed
         self._get_tree_id_for_path(path)
         return self._stat_modes[path]
 
     def is_link(self, path):
         return stat.S_ISLNK(self.get_file_mode(path))
 
     def is_node_binary(self, path):
         tree_id, _ = self._get_tree_id_for_path(path)
         return self._remote.is_binary(tree_id)
 
     def get_file_content(self, path):
         """
         Returns content of the file at given `path`.
         """
         tree_id, _ = self._get_tree_id_for_path(path)
         return self._remote.blob_as_pretty_string(tree_id)
 
     def get_file_content_streamed(self, path):
         tree_id, _ = self._get_tree_id_for_path(path)
         stream_method = getattr(self._remote, 'stream:blob_as_pretty_string')
         return stream_method(tree_id)
 
     def get_file_size(self, path):
         """
         Returns size of the file at given `path`.
         """
         tree_id, _ = self._get_tree_id_for_path(path)
         return self._remote.blob_raw_length(tree_id)
 
     def get_path_history(self, path, limit=None, pre_load=None):
         """
         Returns history of file as reversed list of `GitCommit` objects for
         which file at given `path` has been modified.
         """
 
         path = self._get_filectx(path)
         hist = self._remote.node_history(self.raw_id, path, limit)
         return [
             self.repository.get_commit(commit_id=commit_id, pre_load=pre_load)
             for commit_id in hist]
 
     def get_file_annotate(self, path, pre_load=None):
         """
         Returns a generator of four element tuples with
             lineno, commit_id, commit lazy loader and line
         """
 
         result = self._remote.node_annotate(self.raw_id, path)
 
         for ln_no, commit_id, content in result:
             yield (
                 ln_no, commit_id,
                 lambda: self.repository.get_commit(commit_id=commit_id, pre_load=pre_load),
                 content)
 
     def get_nodes(self, path):
 
         if self._get_kind(path) != NodeKind.DIR:
             raise CommitError(
                 "Directory does not exist for commit %s at '%s'" % (self.raw_id, path))
         path = self._fix_path(path)
 
         tree_id, _ = self._get_tree_id_for_path(path)
 
         dirnodes = []
         filenodes = []
 
         # extracted tree ID gives us our files...
         bytes_path = safe_str(path)  # libgit operates on bytes
         for name, stat_, id_, type_ in self._remote.tree_items(tree_id):
             if type_ == 'link':
                 url = self._get_submodule_url('/'.join((bytes_path, name)))
                 dirnodes.append(SubModuleNode(
                     name, url=url, commit=id_, alias=self.repository.alias))
                 continue
 
             if bytes_path != '':
                 obj_path = '/'.join((bytes_path, name))
             else:
                 obj_path = name
             if obj_path not in self._stat_modes:
                 self._stat_modes[obj_path] = stat_
 
             if type_ == 'tree':
                 dirnodes.append(DirNode(obj_path, commit=self))
             elif type_ == 'blob':
                 filenodes.append(FileNode(obj_path, commit=self, mode=stat_))
             else:
                 raise CommitError(
                     "Requested object should be Tree or Blob, is %s", type_)
 
         nodes = dirnodes + filenodes
         for node in nodes:
             if node.path not in self.nodes:
                 self.nodes[node.path] = node
         nodes.sort()
         return nodes
 
     def get_node(self, path, pre_load=None):
         if isinstance(path, unicode):
             path = path.encode('utf-8')
         path = self._fix_path(path)
         if path not in self.nodes:
             try:
                 tree_id, type_ = self._get_tree_id_for_path(path)
             except CommitError:
                 raise NodeDoesNotExistError(
                     "Cannot find one of parents' directories for a given "
                     "path: %s" % path)
 
             if type_ in ['link', 'commit']:
                 url = self._get_submodule_url(path)
                 node = SubModuleNode(path, url=url, commit=tree_id,
                                      alias=self.repository.alias)
             elif type_ == 'tree':
                 if path == '':
                     node = RootNode(commit=self)
                 else:
                     node = DirNode(path, commit=self)
             elif type_ == 'blob':
                 node = FileNode(path, commit=self, pre_load=pre_load)
                 self._stat_modes[path] = node.mode
             else:
                 raise self.no_node_at_path(path)
 
             # cache node
             self.nodes[path] = node
 
         return self.nodes[path]
 
     def get_largefile_node(self, path):
         tree_id, _ = self._get_tree_id_for_path(path)
         pointer_spec = self._remote.is_large_file(tree_id)
 
         if pointer_spec:
             # content of that file regular FileNode is the hash of largefile
             file_id = pointer_spec.get('oid_hash')
             if self._remote.in_largefiles_store(file_id):
                 lf_path = self._remote.store_path(file_id)
                 return LargeFileNode(lf_path, commit=self, org_path=path)
 
     @LazyProperty
     def affected_files(self):
         """
         Gets a fast accessible file changes for given commit
         """
         added, modified, deleted = self._changes_cache
         return list(added.union(modified).union(deleted))
 
     @LazyProperty
     def _changes_cache(self):
         added = set()
         modified = set()
         deleted = set()
         _r = self._remote
 
         parents = self.parents
         if not self.parents:
             parents = [base.EmptyCommit()]
         for parent in parents:
             if isinstance(parent, base.EmptyCommit):
                 oid = None
             else:
                 oid = parent.raw_id
             changes = _r.tree_changes(oid, self.raw_id)
             for (oldpath, newpath), (_, _), (_, _) in changes:
                 if newpath and oldpath:
                     modified.add(newpath)
                 elif newpath and not oldpath:
                     added.add(newpath)
                 elif not newpath and oldpath:
                     deleted.add(oldpath)
         return added, modified, deleted
 
     def _get_paths_for_status(self, status):
         """
         Returns sorted list of paths for given ``status``.
 
         :param status: one of: *added*, *modified* or *deleted*
         """
         added, modified, deleted = self._changes_cache
         return sorted({
             'added': list(added),
             'modified': list(modified),
             'deleted': list(deleted)}[status]
         )
 
     @LazyProperty
     def added(self):
         """
         Returns list of added ``FileNode`` objects.
         """
         if not self.parents:
             return list(self._get_file_nodes())
         return AddedFileNodesGenerator(self.added_paths, self)
 
     @LazyProperty
     def added_paths(self):
         return [n for n in self._get_paths_for_status('added')]
 
     @LazyProperty
     def changed(self):
         """
         Returns list of modified ``FileNode`` objects.
         """
         if not self.parents:
             return []
         return ChangedFileNodesGenerator(self.changed_paths, self)
 
     @LazyProperty
     def changed_paths(self):
         return [n for n in self._get_paths_for_status('modified')]
 
     @LazyProperty
     def removed(self):
         """
         Returns list of removed ``FileNode`` objects.
         """
         if not self.parents:
             return []
         return RemovedFileNodesGenerator(self.removed_paths, self)
 
     @LazyProperty
     def removed_paths(self):
         return [n for n in self._get_paths_for_status('deleted')]
 
     def _get_submodule_url(self, submodule_path):
         git_modules_path = '.gitmodules'
 
         if self._submodules is None:
             self._submodules = {}
 
             try:
                 submodules_node = self.get_node(git_modules_path)
             except NodeDoesNotExistError:
                 return None
 
             # ConfigParser fails if there are whitespaces, also it needs an iterable
             # file like content
             def iter_content(_content):
                 for line in _content.splitlines():
                     yield line
 
             parser = configparser.RawConfigParser()
             parser.read_file(iter_content(submodules_node.content))
 
             for section in parser.sections():
                 path = parser.get(section, 'path')
                 url = parser.get(section, 'url')
                 if path and url:
                     self._submodules[path.strip('/')] = url
 
         return self._submodules.get(submodule_path.strip('/'))

rhodecode/tests/lib/middleware/test_simplesvn.py

@@ -1,209 +1,209 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
-from StringIO import StringIO
+from io import StringIO
 
 import pytest
 from mock import patch, Mock
 
 from rhodecode.lib.middleware.simplesvn import SimpleSvn, SimpleSvnApp
 from rhodecode.lib.utils import get_rhodecode_base_path
 
 
 class TestSimpleSvn(object):
     @pytest.fixture(autouse=True)
     def simple_svn(self, baseapp, request_stub):
         base_path = get_rhodecode_base_path()
         self.app = SimpleSvn(
             config={'auth_ret_code': '', 'base_path': base_path},
             registry=request_stub.registry)
 
     def test_get_config(self):
         extras = {'foo': 'FOO', 'bar': 'BAR'}
         config = self.app._create_config(extras, repo_name='test-repo')
         assert config == extras
 
     @pytest.mark.parametrize(
         'method', ['OPTIONS', 'PROPFIND', 'GET', 'REPORT'])
     def test_get_action_returns_pull(self, method):
         environment = {'REQUEST_METHOD': method}
         action = self.app._get_action(environment)
         assert action == 'pull'
 
     @pytest.mark.parametrize(
         'method', [
             'MKACTIVITY', 'PROPPATCH', 'PUT', 'CHECKOUT', 'MKCOL', 'MOVE',
             'COPY', 'DELETE', 'LOCK', 'UNLOCK', 'MERGE'
         ])
     def test_get_action_returns_push(self, method):
         environment = {'REQUEST_METHOD': method}
         action = self.app._get_action(environment)
         assert action == 'push'
 
     @pytest.mark.parametrize(
         'path, expected_name', [
             ('/hello-svn', 'hello-svn'),
             ('/hello-svn/', 'hello-svn'),
             ('/group/hello-svn/', 'group/hello-svn'),
             ('/group/hello-svn/!svn/vcc/default', 'group/hello-svn'),
         ])
     def test_get_repository_name(self, path, expected_name):
         environment = {'PATH_INFO': path}
         name = self.app._get_repository_name(environment)
         assert name == expected_name
 
     def test_get_repository_name_subfolder(self, backend_svn):
         repo = backend_svn.repo
         environment = {
             'PATH_INFO': '/{}/path/with/subfolders'.format(repo.repo_name)}
         name = self.app._get_repository_name(environment)
         assert name == repo.repo_name
 
     def test_create_wsgi_app(self):
         with patch.object(SimpleSvn, '_is_svn_enabled') as mock_method:
             mock_method.return_value = False
             with patch('rhodecode.lib.middleware.simplesvn.DisabledSimpleSvnApp') as (
                     wsgi_app_mock):
                 config = Mock()
                 wsgi_app = self.app._create_wsgi_app(
                     repo_path='', repo_name='', config=config)
 
             wsgi_app_mock.assert_called_once_with(config)
             assert wsgi_app == wsgi_app_mock()
 
     def test_create_wsgi_app_when_enabled(self):
         with patch.object(SimpleSvn, '_is_svn_enabled') as mock_method:
             mock_method.return_value = True
             with patch('rhodecode.lib.middleware.simplesvn.SimpleSvnApp') as (
                     wsgi_app_mock):
                 config = Mock()
                 wsgi_app = self.app._create_wsgi_app(
                     repo_path='', repo_name='', config=config)
 
             wsgi_app_mock.assert_called_once_with(config)
             assert wsgi_app == wsgi_app_mock()
 
 
 class TestSimpleSvnApp(object):
     data = '<xml></xml>'
     path = '/group/my-repo'
     wsgi_input = StringIO(data)
     environment = {
         'HTTP_DAV': (
             'http://subversion.tigris.org/xmlns/dav/svn/depth,'
             ' http://subversion.tigris.org/xmlns/dav/svn/mergeinfo'),
         'HTTP_USER_AGENT': 'SVN/1.8.11 (x86_64-linux) serf/1.3.8',
         'REQUEST_METHOD': 'OPTIONS',
         'PATH_INFO': path,
         'wsgi.input': wsgi_input,
         'CONTENT_TYPE': 'text/xml',
         'CONTENT_LENGTH': '130'
     }
 
     def setup_method(self, method):
         self.host = 'http://localhost/'
         base_path = get_rhodecode_base_path()
         self.app = SimpleSvnApp(
             config={'subversion_http_server_url': self.host,
                     'base_path': base_path})
 
     def test_get_request_headers_with_content_type(self):
         expected_headers = {
             'Dav': self.environment['HTTP_DAV'],
             'User-Agent': self.environment['HTTP_USER_AGENT'],
             'Content-Type': self.environment['CONTENT_TYPE'],
             'Content-Length': self.environment['CONTENT_LENGTH']
         }
         headers = self.app._get_request_headers(self.environment)
         assert headers == expected_headers
 
     def test_get_request_headers_without_content_type(self):
         environment = self.environment.copy()
         environment.pop('CONTENT_TYPE')
         expected_headers = {
             'Dav': environment['HTTP_DAV'],
             'Content-Length': self.environment['CONTENT_LENGTH'],
             'User-Agent': environment['HTTP_USER_AGENT'],
         }
         request_headers = self.app._get_request_headers(environment)
         assert request_headers == expected_headers
 
     def test_get_response_headers(self):
         headers = {
             'Connection': 'keep-alive',
             'Keep-Alive': 'timeout=5, max=100',
             'Transfer-Encoding': 'chunked',
             'Content-Encoding': 'gzip',
             'MS-Author-Via': 'DAV',
             'SVN-Supported-Posts': 'create-txn-with-props'
         }
         expected_headers = [
             ('MS-Author-Via', 'DAV'),
             ('SVN-Supported-Posts', 'create-txn-with-props'),
         ]
         response_headers = self.app._get_response_headers(headers)
         assert sorted(response_headers) == sorted(expected_headers)
 
     @pytest.mark.parametrize('svn_http_url, path_info, expected_url', [
         ('http://localhost:8200', '/repo_name', 'http://localhost:8200/repo_name'),
         ('http://localhost:8200///', '/repo_name', 'http://localhost:8200/repo_name'),
         ('http://localhost:8200', '/group/repo_name', 'http://localhost:8200/group/repo_name'),
         ('http://localhost:8200/', '/group/repo_name', 'http://localhost:8200/group/repo_name'),
         ('http://localhost:8200/prefix', '/repo_name', 'http://localhost:8200/prefix/repo_name'),
         ('http://localhost:8200/prefix', 'repo_name', 'http://localhost:8200/prefix/repo_name'),
         ('http://localhost:8200/prefix', '/group/repo_name', 'http://localhost:8200/prefix/group/repo_name')
     ])
     def test_get_url(self, svn_http_url, path_info, expected_url):
         url = self.app._get_url(svn_http_url, path_info)
         assert url == expected_url
 
     def test_call(self):
         start_response = Mock()
         response_mock = Mock()
         response_mock.headers = {
             'Content-Encoding': 'gzip',
             'MS-Author-Via': 'DAV',
             'SVN-Supported-Posts': 'create-txn-with-props'
         }
         response_mock.status_code = 200
         response_mock.reason = 'OK'
         with patch('rhodecode.lib.middleware.simplesvn.requests.request') as (
                 request_mock):
             request_mock.return_value = response_mock
             self.app(self.environment, start_response)
 
         expected_url = '{}{}'.format(self.host.strip('/'), self.path)
         expected_request_headers = {
             'Dav': self.environment['HTTP_DAV'],
             'User-Agent': self.environment['HTTP_USER_AGENT'],
             'Content-Type': self.environment['CONTENT_TYPE'],
             'Content-Length': self.environment['CONTENT_LENGTH']
         }
         expected_response_headers = [
             ('SVN-Supported-Posts', 'create-txn-with-props'),
             ('MS-Author-Via', 'DAV'),
         ]
         request_mock.assert_called_once_with(
             self.environment['REQUEST_METHOD'], expected_url,
             data=self.data, headers=expected_request_headers, stream=False)
         response_mock.iter_content.assert_called_once_with(chunk_size=1024)
         args, _ = start_response.call_args
         assert args[0] == '200 OK'
         assert sorted(args[1]) == sorted(expected_response_headers)

rhodecode/tests/lib/test_hooks_daemon.py

@@ -1,342 +1,342 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import json
 import logging
-from StringIO import StringIO
+from io import StringIO
 
 import mock
 import pytest
 
 from rhodecode.lib import hooks_daemon
 from rhodecode.tests.utils import assert_message_in_log
 
 
 class TestDummyHooksCallbackDaemon(object):
     def test_hooks_module_path_set_properly(self):
         daemon = hooks_daemon.DummyHooksCallbackDaemon()
         assert daemon.hooks_module == 'rhodecode.lib.hooks_daemon'
 
     def test_logs_entering_the_hook(self):
         daemon = hooks_daemon.DummyHooksCallbackDaemon()
         with mock.patch.object(hooks_daemon.log, 'debug') as log_mock:
             with daemon as return_value:
                 log_mock.assert_called_once_with(
                     'Running `%s` callback daemon', 'DummyHooksCallbackDaemon')
         assert return_value == daemon
 
     def test_logs_exiting_the_hook(self):
         daemon = hooks_daemon.DummyHooksCallbackDaemon()
         with mock.patch.object(hooks_daemon.log, 'debug') as log_mock:
             with daemon:
                 pass
         log_mock.assert_called_with(
             'Exiting `%s` callback daemon', 'DummyHooksCallbackDaemon')
 
 
 class TestHooks(object):
     def test_hooks_can_be_used_as_a_context_processor(self):
         hooks = hooks_daemon.Hooks()
         with hooks as return_value:
             pass
         assert hooks == return_value
 
 
 class TestHooksHttpHandler(object):
     def test_read_request_parses_method_name_and_arguments(self):
         data = {
             'method': 'test',
             'extras': {
                 'param1': 1,
                 'param2': 'a'
             }
         }
         request = self._generate_post_request(data)
         hooks_patcher = mock.patch.object(
             hooks_daemon.Hooks, data['method'], create=True, return_value=1)
 
         with hooks_patcher as hooks_mock:
             MockServer(hooks_daemon.HooksHttpHandler, request)
 
         hooks_mock.assert_called_once_with(data['extras'])
 
     def test_hooks_serialized_result_is_returned(self):
         request = self._generate_post_request({})
         rpc_method = 'test'
         hook_result = {
             'first': 'one',
             'second': 2
         }
         read_patcher = mock.patch.object(
             hooks_daemon.HooksHttpHandler, '_read_request',
             return_value=(rpc_method, {}))
         hooks_patcher = mock.patch.object(
             hooks_daemon.Hooks, rpc_method, create=True,
             return_value=hook_result)
 
         with read_patcher, hooks_patcher:
             server = MockServer(hooks_daemon.HooksHttpHandler, request)
 
         expected_result = json.dumps(hook_result)
         assert server.request.output_stream.buflist[-1] == expected_result
 
     def test_exception_is_returned_in_response(self):
         request = self._generate_post_request({})
         rpc_method = 'test'
         read_patcher = mock.patch.object(
             hooks_daemon.HooksHttpHandler, '_read_request',
             return_value=(rpc_method, {}))
         hooks_patcher = mock.patch.object(
             hooks_daemon.Hooks, rpc_method, create=True,
             side_effect=Exception('Test exception'))
 
         with read_patcher, hooks_patcher:
             server = MockServer(hooks_daemon.HooksHttpHandler, request)
 
         org_exc = json.loads(server.request.output_stream.buflist[-1])
         expected_result = {
             'exception': 'Exception',
             'exception_traceback': org_exc['exception_traceback'],
             'exception_args': ['Test exception']
         }
         assert org_exc == expected_result
 
     def test_log_message_writes_to_debug_log(self, caplog):
         ip_port = ('0.0.0.0', 8888)
         handler = hooks_daemon.HooksHttpHandler(
             MockRequest('POST /'), ip_port, mock.Mock())
         fake_date = '1/Nov/2015 00:00:00'
         date_patcher = mock.patch.object(
             handler, 'log_date_time_string', return_value=fake_date)
         with date_patcher, caplog.at_level(logging.DEBUG):
             handler.log_message('Some message %d, %s', 123, 'string')
 
         expected_message = "HOOKS: {} - - [{}] Some message 123, string".format(ip_port, fake_date)
         assert_message_in_log(
             caplog.records, expected_message,
             levelno=logging.DEBUG, module='hooks_daemon')
 
     def _generate_post_request(self, data):
         payload = json.dumps(data)
         return 'POST / HTTP/1.0\nContent-Length: {}\n\n{}'.format(
             len(payload), payload)
 
 
 class ThreadedHookCallbackDaemon(object):
     def test_constructor_calls_prepare(self):
         prepare_daemon_patcher = mock.patch.object(
             hooks_daemon.ThreadedHookCallbackDaemon, '_prepare')
         with prepare_daemon_patcher as prepare_daemon_mock:
             hooks_daemon.ThreadedHookCallbackDaemon()
         prepare_daemon_mock.assert_called_once_with()
 
     def test_run_is_called_on_context_start(self):
         patchers = mock.patch.multiple(
             hooks_daemon.ThreadedHookCallbackDaemon,
             _run=mock.DEFAULT, _prepare=mock.DEFAULT, __exit__=mock.DEFAULT)
 
         with patchers as mocks:
             daemon = hooks_daemon.ThreadedHookCallbackDaemon()
             with daemon as daemon_context:
                 pass
             mocks['_run'].assert_called_once_with()
         assert daemon_context == daemon
 
     def test_stop_is_called_on_context_exit(self):
         patchers = mock.patch.multiple(
             hooks_daemon.ThreadedHookCallbackDaemon,
             _run=mock.DEFAULT, _prepare=mock.DEFAULT, _stop=mock.DEFAULT)
 
         with patchers as mocks:
             daemon = hooks_daemon.ThreadedHookCallbackDaemon()
             with daemon as daemon_context:
                 assert mocks['_stop'].call_count == 0
 
             mocks['_stop'].assert_called_once_with()
         assert daemon_context == daemon
 
 
 class TestHttpHooksCallbackDaemon(object):
     def test_hooks_callback_generates_new_port(self, caplog):
         with caplog.at_level(logging.DEBUG):
             daemon = hooks_daemon.HttpHooksCallbackDaemon(host='127.0.0.1', port=8881)
         assert daemon._daemon.server_address == ('127.0.0.1', 8881)
 
         with caplog.at_level(logging.DEBUG):
             daemon = hooks_daemon.HttpHooksCallbackDaemon(host=None, port=None)
         assert daemon._daemon.server_address[1] in range(0, 66000)
         assert daemon._daemon.server_address[0] != '127.0.0.1'
 
     def test_prepare_inits_daemon_variable(self, tcp_server, caplog):
         with self._tcp_patcher(tcp_server), caplog.at_level(logging.DEBUG):
             daemon = hooks_daemon.HttpHooksCallbackDaemon(host='127.0.0.1', port=8881)
         assert daemon._daemon == tcp_server
 
         _, port = tcp_server.server_address
         expected_uri = '{}:{}'.format('127.0.0.1', port)
         msg = 'HOOKS: {}  Preparing HTTP callback daemon registering ' \
               'hook object: rhodecode.lib.hooks_daemon.HooksHttpHandler'.format(expected_uri)
         assert_message_in_log(
             caplog.records, msg, levelno=logging.DEBUG, module='hooks_daemon')
 
     def test_prepare_inits_hooks_uri_and_logs_it(
             self, tcp_server, caplog):
         with self._tcp_patcher(tcp_server), caplog.at_level(logging.DEBUG):
             daemon = hooks_daemon.HttpHooksCallbackDaemon(host='127.0.0.1', port=8881)
 
         _, port = tcp_server.server_address
         expected_uri = '{}:{}'.format('127.0.0.1', port)
         assert daemon.hooks_uri == expected_uri
 
         msg = 'HOOKS: {}  Preparing HTTP callback daemon registering ' \
               'hook object: rhodecode.lib.hooks_daemon.HooksHttpHandler'.format(expected_uri)
         assert_message_in_log(
             caplog.records, msg,
             levelno=logging.DEBUG, module='hooks_daemon')
 
     def test_run_creates_a_thread(self, tcp_server):
         thread = mock.Mock()
 
         with self._tcp_patcher(tcp_server):
             daemon = hooks_daemon.HttpHooksCallbackDaemon()
 
         with self._thread_patcher(thread) as thread_mock:
             daemon._run()
 
         thread_mock.assert_called_once_with(
             target=tcp_server.serve_forever,
             kwargs={'poll_interval': daemon.POLL_INTERVAL})
         assert thread.daemon is True
         thread.start.assert_called_once_with()
 
     def test_run_logs(self, tcp_server, caplog):
 
         with self._tcp_patcher(tcp_server):
             daemon = hooks_daemon.HttpHooksCallbackDaemon()
 
         with self._thread_patcher(mock.Mock()), caplog.at_level(logging.DEBUG):
             daemon._run()
 
         assert_message_in_log(
             caplog.records,
             'Running event loop of callback daemon in background thread',
             levelno=logging.DEBUG, module='hooks_daemon')
 
     def test_stop_cleans_up_the_connection(self, tcp_server, caplog):
         thread = mock.Mock()
 
         with self._tcp_patcher(tcp_server):
             daemon = hooks_daemon.HttpHooksCallbackDaemon()
 
         with self._thread_patcher(thread), caplog.at_level(logging.DEBUG):
             with daemon:
                 assert daemon._daemon == tcp_server
                 assert daemon._callback_thread == thread
 
         assert daemon._daemon is None
         assert daemon._callback_thread is None
         tcp_server.shutdown.assert_called_with()
         thread.join.assert_called_once_with()
 
         assert_message_in_log(
             caplog.records, 'Waiting for background thread to finish.',
             levelno=logging.DEBUG, module='hooks_daemon')
 
     def _tcp_patcher(self, tcp_server):
         return mock.patch.object(
             hooks_daemon, 'TCPServer', return_value=tcp_server)
 
     def _thread_patcher(self, thread):
         return mock.patch.object(
             hooks_daemon.threading, 'Thread', return_value=thread)
 
 
 class TestPrepareHooksDaemon(object):
     @pytest.mark.parametrize('protocol', ('http',))
     def test_returns_dummy_hooks_callback_daemon_when_using_direct_calls(
             self, protocol):
         expected_extras = {'extra1': 'value1'}
         callback, extras = hooks_daemon.prepare_callback_daemon(
             expected_extras.copy(), protocol=protocol,
             host='127.0.0.1', use_direct_calls=True)
         assert isinstance(callback, hooks_daemon.DummyHooksCallbackDaemon)
         expected_extras['hooks_module'] = 'rhodecode.lib.hooks_daemon'
         expected_extras['time'] = extras['time']
         assert 'extra1' in extras
 
     @pytest.mark.parametrize('protocol, expected_class', (
         ('http', hooks_daemon.HttpHooksCallbackDaemon),
     ))
     def test_returns_real_hooks_callback_daemon_when_protocol_is_specified(
             self, protocol, expected_class):
         expected_extras = {
             'extra1': 'value1',
             'txn_id': 'txnid2',
             'hooks_protocol': protocol.lower()
         }
         callback, extras = hooks_daemon.prepare_callback_daemon(
             expected_extras.copy(), protocol=protocol, host='127.0.0.1',
             use_direct_calls=False,
             txn_id='txnid2')
         assert isinstance(callback, expected_class)
         extras.pop('hooks_uri')
         expected_extras['time'] = extras['time']
         assert extras == expected_extras
 
     @pytest.mark.parametrize('protocol', (
         'invalid',
         'Http',
         'HTTP',
     ))
     def test_raises_on_invalid_protocol(self, protocol):
         expected_extras = {
             'extra1': 'value1',
             'hooks_protocol': protocol.lower()
         }
         with pytest.raises(Exception):
             callback, extras = hooks_daemon.prepare_callback_daemon(
                 expected_extras.copy(),
                 protocol=protocol, host='127.0.0.1',
                 use_direct_calls=False)
 
 
 class MockRequest(object):
     def __init__(self, request):
         self.request = request
         self.input_stream = StringIO(b'{}'.format(self.request))
         self.output_stream = StringIO()
 
     def makefile(self, mode, *args, **kwargs):
         return self.output_stream if mode == 'wb' else self.input_stream
 
 
 class MockServer(object):
     def __init__(self, handler_cls, request):
         ip_port = ('0.0.0.0', 8888)
         self.request = MockRequest(request)
         self.server_address = ip_port
         self.handler = handler_cls(self.request, ip_port, self)
 
 
 @pytest.fixture()
 def tcp_server():
     server = mock.Mock()
     server.server_address = ('127.0.0.1', 8881)
     return server

rhodecode/tests/vcs/test_archives.py

@@ -1,176 +1,176 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import datetime
 import os
 import shutil
 import tarfile
 import tempfile
 import zipfile
-import StringIO
+from io import StringIO
 
 import mock
 import pytest
 
 from rhodecode.lib.vcs.backends import base
 from rhodecode.lib.vcs.exceptions import ImproperArchiveTypeError, VCSError
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.tests.vcs.conftest import BackendTestMixin
 
 
 @pytest.mark.usefixtures("vcs_repository_support")
 class TestArchives(BackendTestMixin):
 
     @pytest.fixture(autouse=True)
     def tempfile(self, request):
         self.temp_file = tempfile.mkstemp()[1]
 
         @request.addfinalizer
         def cleanup():
             os.remove(self.temp_file)
 
     @classmethod
     def _get_commits(cls):
         start_date = datetime.datetime(2010, 1, 1, 20)
         yield {
             'message': 'Initial Commit',
             'author': 'Joe Doe <[e-mail]>',
             'date': start_date + datetime.timedelta(hours=12),
             'added': [
                 FileNode('executable_0o100755', '...', mode=0o100755),
                 FileNode('executable_0o100500', '...', mode=0o100500),
                 FileNode('not_executable', '...', mode=0o100644),
             ],
         }
         for x in range(5):
             yield {
                 'message': 'Commit %d' % x,
                 'author': 'Joe Doe <[e-mail]>',
                 'date': start_date + datetime.timedelta(hours=12 * x),
                 'added': [
                     FileNode('%d/file_%d.txt' % (x, x), content='Foobar %d' % x),
                 ],
             }
 
     @pytest.mark.parametrize('compressor', ['gz', 'bz2'])
     def test_archive_tar(self, compressor):
         self.tip.archive_repo(
             self.temp_file, kind='t{}'.format(compressor), archive_dir_name='repo')
         out_dir = tempfile.mkdtemp()
         out_file = tarfile.open(self.temp_file, 'r|{}'.format(compressor))
         out_file.extractall(out_dir)
         out_file.close()
 
         for x in range(5):
             node_path = '%d/file_%d.txt' % (x, x)
             with open(os.path.join(out_dir, 'repo/' + node_path)) as f:
                 file_content = f.read()
             assert file_content == self.tip.get_node(node_path).content
 
         shutil.rmtree(out_dir)
 
     @pytest.mark.parametrize('compressor', ['gz', 'bz2'])
     def test_archive_tar_symlink(self, compressor):
         return False
 
     @pytest.mark.parametrize('compressor', ['gz', 'bz2'])
     def test_archive_tar_file_modes(self, compressor):
         self.tip.archive_repo(
             self.temp_file, kind='t{}'.format(compressor), archive_dir_name='repo')
         out_dir = tempfile.mkdtemp()
         out_file = tarfile.open(self.temp_file, 'r|{}'.format(compressor))
         out_file.extractall(out_dir)
         out_file.close()
         dest = lambda inp: os.path.join(out_dir, 'repo/' + inp)
 
         assert oct(os.stat(dest('not_executable')).st_mode) == '0100644'
 
     def test_archive_zip(self):
         self.tip.archive_repo(self.temp_file, kind='zip', archive_dir_name='repo')
         out = zipfile.ZipFile(self.temp_file)
 
         for x in range(5):
             node_path = '%d/file_%d.txt' % (x, x)
             decompressed = StringIO.StringIO()
             decompressed.write(out.read('repo/' + node_path))
             assert decompressed.getvalue() == \
                 self.tip.get_node(node_path).content
             decompressed.close()
 
     def test_archive_zip_with_metadata(self):
         self.tip.archive_repo(self.temp_file, kind='zip',
                               archive_dir_name='repo', write_metadata=True)
 
         out = zipfile.ZipFile(self.temp_file)
         metafile = out.read('repo/.archival.txt')
 
         raw_id = self.tip.raw_id
         assert 'commit_id:%s' % raw_id in metafile
 
         for x in range(5):
             node_path = '%d/file_%d.txt' % (x, x)
             decompressed = StringIO.StringIO()
             decompressed.write(out.read('repo/' + node_path))
             assert decompressed.getvalue() == \
                 self.tip.get_node(node_path).content
             decompressed.close()
 
     def test_archive_wrong_kind(self):
         with pytest.raises(ImproperArchiveTypeError):
             self.tip.archive_repo(self.temp_file, kind='wrong kind')
 
 
 @pytest.fixture()
 def base_commit():
     """
     Prepare a `base.BaseCommit` just enough for `_validate_archive_prefix`.
     """
     commit = base.BaseCommit()
     commit.repository = mock.Mock()
     commit.repository.name = u'fake_repo'
     commit.short_id = 'fake_id'
     return commit
 
 
 @pytest.mark.parametrize("prefix", [u"unicode-prefix", u"Ünïcödë"])
 def test_validate_archive_prefix_enforces_bytes_as_prefix(prefix, base_commit):
     with pytest.raises(ValueError):
         base_commit._validate_archive_prefix(prefix)
 
 
 def test_validate_archive_prefix_empty_prefix(base_commit):
     # TODO: johbo: Should raise a ValueError here.
     with pytest.raises(VCSError):
         base_commit._validate_archive_prefix('')
 
 
 def test_validate_archive_prefix_with_leading_slash(base_commit):
     # TODO: johbo: Should raise a ValueError here.
     with pytest.raises(VCSError):
         base_commit._validate_archive_prefix('/any')
 
 
 def test_validate_archive_prefix_falls_back_to_repository_name(base_commit):
     prefix = base_commit._validate_archive_prefix(None)
     expected_prefix = base_commit._ARCHIVE_PREFIX_TEMPLATE.format(
         repo_name='fake_repo',
         short_id='fake_id')
     assert isinstance(prefix, str)
     assert prefix == expected_prefix