helpers: remove usage of pylons session.

rhodecode/apps/admin/tests/test_admin_defaults.py

@@ -1,85 +1,85 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 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 pytest
 
 from rhodecode.tests import assert_session_flash
 from rhodecode.model.settings import SettingsModel
 
 
 def route_path(name, params=None, **kwargs):
     import urllib
     from rhodecode.apps._base import ADMIN_PREFIX
 
     base_url = {
         'admin_defaults_repositories':
             ADMIN_PREFIX + '/defaults/repositories',
         'admin_defaults_repositories_update':
             ADMIN_PREFIX + '/defaults/repositories/update',
     }[name].format(**kwargs)
 
     if params:
         base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
     return base_url
 
 
 @pytest.mark.usefixtures("app")
-class TestDefaultsController(object):
+class TestDefaultsView(object):
 
     def test_index(self, autologin_user):
         response = self.app.get(route_path('admin_defaults_repositories'))
         response.mustcontain('default_repo_private')
         response.mustcontain('default_repo_enable_statistics')
         response.mustcontain('default_repo_enable_downloads')
         response.mustcontain('default_repo_enable_locking')
 
     def test_update_params_true_hg(self, autologin_user, csrf_token):
         params = {
             'default_repo_enable_locking': True,
             'default_repo_enable_downloads': True,
             'default_repo_enable_statistics': True,
             'default_repo_private': True,
             'default_repo_type': 'hg',
             'csrf_token': csrf_token,
         }
         response = self.app.post(
             route_path('admin_defaults_repositories_update'), params=params)
         assert_session_flash(response, 'Default settings updated successfully')
 
         defs = SettingsModel().get_default_repo_settings()
         del params['csrf_token']
         assert params == defs
 
     def test_update_params_false_git(self, autologin_user, csrf_token):
         params = {
             'default_repo_enable_locking': False,
             'default_repo_enable_downloads': False,
             'default_repo_enable_statistics': False,
             'default_repo_private': False,
             'default_repo_type': 'git',
             'csrf_token': csrf_token,
         }
         response = self.app.post(
             route_path('admin_defaults_repositories_update'), params=params)
         assert_session_flash(response, 'Default settings updated successfully')
 
         defs = SettingsModel().get_default_repo_settings()
         del params['csrf_token']
         assert params == defs

rhodecode/lib/base.py

@@ -1,617 +1,627 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 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/
 
 """
 The base Controller API
 Provides the BaseController class for subclassing. And usage in different
 controllers
 """
 
 import logging
 import socket
 
 import markupsafe
 import ipaddress
 import pyramid.threadlocal
 
 from paste.auth.basic import AuthBasicAuthenticator
 from paste.httpexceptions import HTTPUnauthorized, HTTPForbidden, get_exception
 from paste.httpheaders import WWW_AUTHENTICATE, AUTHORIZATION
 
 import rhodecode
 from rhodecode.authentication.base import VCS_TYPE
 from rhodecode.lib import auth, utils2
 from rhodecode.lib import helpers as h
 from rhodecode.lib.auth import AuthUser, CookieStoreWrapper
 from rhodecode.lib.exceptions import UserCreationError
 from rhodecode.lib.utils import (
     get_repo_slug, set_rhodecode_config, password_changed,
     get_enabled_hook_classes)
 from rhodecode.lib.utils2 import (
     str2bool, safe_unicode, AttributeDict, safe_int, md5, aslist)
 from rhodecode.model import meta
 from rhodecode.model.db import Repository, User, ChangesetComment
 from rhodecode.model.notification import NotificationModel
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.settings import VcsSettingsModel, SettingsModel
 
 # NOTE(marcink): remove after base controller is no longer required
 from pylons.controllers import WSGIController
 from pylons.i18n import translation
 
 log = logging.getLogger(__name__)
 
 
 # hack to make the migration to pyramid easier
 def render(template_name, extra_vars=None, cache_key=None,
            cache_type=None, cache_expire=None):
     """Render a template with Mako
 
     Accepts the cache options ``cache_key``, ``cache_type``, and
     ``cache_expire``.
 
     """
     from pylons.templating import literal
     from pylons.templating import cached_template, pylons_globals
 
     # Create a render callable for the cache function
     def render_template():
         # Pull in extra vars if needed
         globs = extra_vars or {}
 
         # Second, get the globals
         globs.update(pylons_globals())
 
         globs['_ungettext'] = globs['ungettext']
         # Grab a template reference
         template = globs['app_globals'].mako_lookup.get_template(template_name)
 
         return literal(template.render_unicode(**globs))
 
     return cached_template(template_name, render_template, cache_key=cache_key,
                            cache_type=cache_type, cache_expire=cache_expire)
 
 def _filter_proxy(ip):
     """
     Passed in IP addresses in HEADERS can be in a special format of multiple
     ips. Those comma separated IPs are passed from various proxies in the
     chain of request processing. The left-most being the original client.
     We only care about the first IP which came from the org. client.
 
     :param ip: ip string from headers
     """
     if ',' in ip:
         _ips = ip.split(',')
         _first_ip = _ips[0].strip()
         log.debug('Got multiple IPs %s, using %s', ','.join(_ips), _first_ip)
         return _first_ip
     return ip
 
 
 def _filter_port(ip):
     """
     Removes a port from ip, there are 4 main cases to handle here.
     - ipv4 eg. 127.0.0.1
     - ipv6 eg. ::1
     - ipv4+port eg. 127.0.0.1:8080
     - ipv6+port eg. [::1]:8080
 
     :param ip:
     """
     def is_ipv6(ip_addr):
         if hasattr(socket, 'inet_pton'):
             try:
                 socket.inet_pton(socket.AF_INET6, ip_addr)
             except socket.error:
                 return False
         else:
             # fallback to ipaddress
             try:
                 ipaddress.IPv6Address(safe_unicode(ip_addr))
             except Exception:
                 return False
         return True
 
     if ':' not in ip:  # must be ipv4 pure ip
         return ip
 
     if '[' in ip and ']' in ip:  # ipv6 with port
         return ip.split(']')[0][1:].lower()
 
     # must be ipv6 or ipv4 with port
     if is_ipv6(ip):
         return ip
     else:
         ip, _port = ip.split(':')[:2]  # means ipv4+port
         return ip
 
 
 def get_ip_addr(environ):
     proxy_key = 'HTTP_X_REAL_IP'
     proxy_key2 = 'HTTP_X_FORWARDED_FOR'
     def_key = 'REMOTE_ADDR'
     _filters = lambda x: _filter_port(_filter_proxy(x))
 
     ip = environ.get(proxy_key)
     if ip:
         return _filters(ip)
 
     ip = environ.get(proxy_key2)
     if ip:
         return _filters(ip)
 
     ip = environ.get(def_key, '0.0.0.0')
     return _filters(ip)
 
 
 def get_server_ip_addr(environ, log_errors=True):
     hostname = environ.get('SERVER_NAME')
     try:
         return socket.gethostbyname(hostname)
     except Exception as e:
         if log_errors:
             # in some cases this lookup is not possible, and we don't want to
             # make it an exception in logs
             log.exception('Could not retrieve server ip address: %s', e)
         return hostname
 
 
 def get_server_port(environ):
     return environ.get('SERVER_PORT')
 
 
 def get_access_path(environ):
     path = environ.get('PATH_INFO')
     org_req = environ.get('pylons.original_request')
     if org_req:
         path = org_req.environ.get('PATH_INFO')
     return path
 
 
 def get_user_agent(environ):
     return environ.get('HTTP_USER_AGENT')
 
 
 def vcs_operation_context(
         environ, repo_name, username, action, scm, check_locking=True,
         is_shadow_repo=False):
     """
     Generate the context for a vcs operation, e.g. push or pull.
 
     This context is passed over the layers so that hooks triggered by the
     vcs operation know details like the user, the user's IP address etc.
 
     :param check_locking: Allows to switch of the computation of the locking
         data. This serves mainly the need of the simplevcs middleware to be
         able to disable this for certain operations.
 
     """
     # Tri-state value: False: unlock, None: nothing, True: lock
     make_lock = None
     locked_by = [None, None, None]
     is_anonymous = username == User.DEFAULT_USER
     if not is_anonymous and check_locking:
         log.debug('Checking locking on repository "%s"', repo_name)
         user = User.get_by_username(username)
         repo = Repository.get_by_repo_name(repo_name)
         make_lock, __, locked_by = repo.get_locking_state(
             action, user.user_id)
 
     settings_model = VcsSettingsModel(repo=repo_name)
     ui_settings = settings_model.get_ui_settings()
 
     extras = {
         'ip': get_ip_addr(environ),
         'username': username,
         'action': action,
         'repository': repo_name,
         'scm': scm,
         'config': rhodecode.CONFIG['__file__'],
         'make_lock': make_lock,
         'locked_by': locked_by,
         'server_url': utils2.get_server_url(environ),
         'user_agent': get_user_agent(environ),
         'hooks': get_enabled_hook_classes(ui_settings),
         'is_shadow_repo': is_shadow_repo,
     }
     return extras
 
 
 class BasicAuth(AuthBasicAuthenticator):
 
     def __init__(self, realm, authfunc, registry, auth_http_code=None,
                  initial_call_detection=False, acl_repo_name=None):
         self.realm = realm
         self.initial_call = initial_call_detection
         self.authfunc = authfunc
         self.registry = registry
         self.acl_repo_name = acl_repo_name
         self._rc_auth_http_code = auth_http_code
 
     def _get_response_from_code(self, http_code):
         try:
             return get_exception(safe_int(http_code))
         except Exception:
             log.exception('Failed to fetch response for code %s' % http_code)
             return HTTPForbidden
 
     def build_authentication(self):
         head = WWW_AUTHENTICATE.tuples('Basic realm="%s"' % self.realm)
         if self._rc_auth_http_code and not self.initial_call:
             # return alternative HTTP code if alternative http return code
             # is specified in RhodeCode config, but ONLY if it's not the
             # FIRST call
             custom_response_klass = self._get_response_from_code(
                 self._rc_auth_http_code)
             return custom_response_klass(headers=head)
         return HTTPUnauthorized(headers=head)
 
     def authenticate(self, environ):
         authorization = AUTHORIZATION(environ)
         if not authorization:
             return self.build_authentication()
         (authmeth, auth) = authorization.split(' ', 1)
         if 'basic' != authmeth.lower():
             return self.build_authentication()
         auth = auth.strip().decode('base64')
         _parts = auth.split(':', 1)
         if len(_parts) == 2:
             username, password = _parts
             if self.authfunc(
                     username, password, environ, VCS_TYPE,
                     registry=self.registry, acl_repo_name=self.acl_repo_name):
                 return username
             if username and password:
                 # we mark that we actually executed authentication once, at
                 # that point we can use the alternative auth code
                 self.initial_call = False
 
         return self.build_authentication()
 
     __call__ = authenticate
 
 
 def calculate_version_hash(config):
     return md5(
         config.get('beaker.session.secret', '') +
         rhodecode.__version__)[:8]
 
 
 def get_current_lang(request):
     # NOTE(marcink): remove after pyramid move
     try:
         return translation.get_lang()[0]
     except:
         pass
 
     return getattr(request, '_LOCALE_', request.locale_name)
 
 
 def attach_context_attributes(context, request, user_id):
     """
     Attach variables into template context called `c`, please note that
     request could be pylons or pyramid request in here.
     """
     # NOTE(marcink): remove check after pyramid migration
     if hasattr(request, 'registry'):
         config = request.registry.settings
     else:
         from pylons import config
 
     rc_config = SettingsModel().get_all_settings(cache=True)
 
     context.rhodecode_version = rhodecode.__version__
     context.rhodecode_edition = config.get('rhodecode.edition')
     # unique secret + version does not leak the version but keep consistency
     context.rhodecode_version_hash = calculate_version_hash(config)
 
     # Default language set for the incoming request
     context.language = get_current_lang(request)
 
     # Visual options
     context.visual = AttributeDict({})
 
     # DB stored Visual Items
     context.visual.show_public_icon = str2bool(
         rc_config.get('rhodecode_show_public_icon'))
     context.visual.show_private_icon = str2bool(
         rc_config.get('rhodecode_show_private_icon'))
     context.visual.stylify_metatags = str2bool(
         rc_config.get('rhodecode_stylify_metatags'))
     context.visual.dashboard_items = safe_int(
         rc_config.get('rhodecode_dashboard_items', 100))
     context.visual.admin_grid_items = safe_int(
         rc_config.get('rhodecode_admin_grid_items', 100))
     context.visual.repository_fields = str2bool(
         rc_config.get('rhodecode_repository_fields'))
     context.visual.show_version = str2bool(
         rc_config.get('rhodecode_show_version'))
     context.visual.use_gravatar = str2bool(
         rc_config.get('rhodecode_use_gravatar'))
     context.visual.gravatar_url = rc_config.get('rhodecode_gravatar_url')
     context.visual.default_renderer = rc_config.get(
         'rhodecode_markup_renderer', 'rst')
     context.visual.comment_types = ChangesetComment.COMMENT_TYPES
     context.visual.rhodecode_support_url = \
         rc_config.get('rhodecode_support_url') or h.route_url('rhodecode_support')
 
     context.visual.affected_files_cut_off = 60
 
     context.pre_code = rc_config.get('rhodecode_pre_code')
     context.post_code = rc_config.get('rhodecode_post_code')
     context.rhodecode_name = rc_config.get('rhodecode_title')
     context.default_encodings = aslist(config.get('default_encoding'), sep=',')
     # if we have specified default_encoding in the request, it has more
     # priority
     if request.GET.get('default_encoding'):
         context.default_encodings.insert(0, request.GET.get('default_encoding'))
     context.clone_uri_tmpl = rc_config.get('rhodecode_clone_uri_tmpl')
 
     # INI stored
     context.labs_active = str2bool(
         config.get('labs_settings_active', 'false'))
     context.visual.allow_repo_location_change = str2bool(
         config.get('allow_repo_location_change', True))
     context.visual.allow_custom_hooks_settings = str2bool(
         config.get('allow_custom_hooks_settings', True))
     context.debug_style = str2bool(config.get('debug_style', False))
 
     context.rhodecode_instanceid = config.get('instance_id')
 
     context.visual.cut_off_limit_diff = safe_int(
         config.get('cut_off_limit_diff'))
     context.visual.cut_off_limit_file = safe_int(
         config.get('cut_off_limit_file'))
 
     # AppEnlight
     context.appenlight_enabled = str2bool(config.get('appenlight', 'false'))
     context.appenlight_api_public_key = config.get(
         'appenlight.api_public_key', '')
     context.appenlight_server_url = config.get('appenlight.server_url', '')
 
     # JS template context
     context.template_context = {
         'repo_name': None,
         'repo_type': None,
         'repo_landing_commit': None,
         'rhodecode_user': {
             'username': None,
             'email': None,
             'notification_status': False
         },
         'visual': {
             'default_renderer': None
         },
         'commit_data': {
             'commit_id': None
         },
         'pull_request_data': {'pull_request_id': None},
         'timeago': {
             'refresh_time': 120 * 1000,
             'cutoff_limit': 1000 * 60 * 60 * 24 * 7
         },
         'pyramid_dispatch': {
 
         },
         'extra': {'plugins': {}}
     }
     # END CONFIG VARS
 
     # TODO: This dosn't work when called from pylons compatibility tween.
     # Fix this and remove it from base controller.
     # context.repo_name = get_repo_slug(request)  # can be empty
 
     diffmode = 'sideside'
     if request.GET.get('diffmode'):
         if request.GET['diffmode'] == 'unified':
             diffmode = 'unified'
     elif request.session.get('diffmode'):
         diffmode = request.session['diffmode']
 
     context.diffmode = diffmode
 
     if request.session.get('diffmode') != diffmode:
         request.session['diffmode'] = diffmode
 
     context.csrf_token = auth.get_csrf_token(session=request.session)
     context.backends = rhodecode.BACKENDS.keys()
     context.backends.sort()
     context.unread_notifications = NotificationModel().get_unread_cnt_for_user(user_id)
 
     # NOTE(marcink): when migrated to pyramid we don't need to set this anymore,
     # given request will ALWAYS be pyramid one
     pyramid_request = pyramid.threadlocal.get_current_request()
     context.pyramid_request = pyramid_request
 
     # web case
     if hasattr(pyramid_request, 'user'):
         context.auth_user = pyramid_request.user
         context.rhodecode_user = pyramid_request.user
 
     # api case
     if hasattr(pyramid_request, 'rpc_user'):
         context.auth_user = pyramid_request.rpc_user
         context.rhodecode_user = pyramid_request.rpc_user
 
     # attach the whole call context to the request
     request.call_context = context
 
 
 def get_auth_user(request):
     environ = request.environ
     session = request.session
 
     ip_addr = get_ip_addr(environ)
     # make sure that we update permissions each time we call controller
     _auth_token = (request.GET.get('auth_token', '') or
                    request.GET.get('api_key', ''))
 
     if _auth_token:
         # when using API_KEY we assume user exists, and
         # doesn't need auth based on cookies.
         auth_user = AuthUser(api_key=_auth_token, ip_addr=ip_addr)
         authenticated = False
     else:
         cookie_store = CookieStoreWrapper(session.get('rhodecode_user'))
         try:
             auth_user = AuthUser(user_id=cookie_store.get('user_id', None),
                                  ip_addr=ip_addr)
         except UserCreationError as e:
             h.flash(e, 'error')
             # container auth or other auth functions that create users
             # on the fly can throw this exception signaling that there's
             # issue with user creation, explanation should be provided
             # in Exception itself. We then create a simple blank
             # AuthUser
             auth_user = AuthUser(ip_addr=ip_addr)
 
         if password_changed(auth_user, session):
             session.invalidate()
             cookie_store = CookieStoreWrapper(session.get('rhodecode_user'))
             auth_user = AuthUser(ip_addr=ip_addr)
 
         authenticated = cookie_store.get('is_authenticated')
 
     if not auth_user.is_authenticated and auth_user.is_user_object:
         # user is not authenticated and not empty
         auth_user.set_authenticated(authenticated)
 
     return auth_user
 
 
 class BaseController(WSGIController):
 
     def __before__(self):
         """
         __before__ is called before controller methods and after __call__
         """
         # on each call propagate settings calls into global settings.
         from pylons import config
         from pylons import tmpl_context as c, request, url
         set_rhodecode_config(config)
         attach_context_attributes(c, request, self._rhodecode_user.user_id)
 
         # TODO: Remove this when fixed in attach_context_attributes()
         c.repo_name = get_repo_slug(request)  # can be empty
 
         self.cut_off_limit_diff = safe_int(config.get('cut_off_limit_diff'))
         self.cut_off_limit_file = safe_int(config.get('cut_off_limit_file'))
         self.sa = meta.Session
         self.scm_model = ScmModel(self.sa)
 
         # set user language
         user_lang = getattr(c.pyramid_request, '_LOCALE_', None)
         if user_lang:
             translation.set_lang(user_lang)
             log.debug('set language to %s for user %s',
                       user_lang, self._rhodecode_user)
 
     def _dispatch_redirect(self, with_url, environ, start_response):
         from webob.exc import HTTPFound
         resp = HTTPFound(with_url)
         environ['SCRIPT_NAME'] = ''  # handle prefix middleware
         environ['PATH_INFO'] = with_url
         return resp(environ, start_response)
 
     def __call__(self, environ, start_response):
         """Invoke the Controller"""
         # WSGIController.__call__ dispatches to the Controller method
         # the request is routed to. This routing information is
         # available in environ['pylons.routes_dict']
         from rhodecode.lib import helpers as h
         from pylons import tmpl_context as c, request, url
 
         # Provide the Pylons context to Pyramid's debugtoolbar if it asks
         if environ.get('debugtoolbar.wants_pylons_context', False):
             environ['debugtoolbar.pylons_context'] = c._current_obj()
 
         _route_name = '.'.join([environ['pylons.routes_dict']['controller'],
                                 environ['pylons.routes_dict']['action']])
 
         self.rc_config = SettingsModel().get_all_settings(cache=True)
         self.ip_addr = get_ip_addr(environ)
 
         # The rhodecode auth user is looked up and passed through the
         # environ by the pylons compatibility tween in pyramid.
         # So we can just grab it from there.
         auth_user = environ['rc_auth_user']
 
         # set globals for auth user
         request.user = auth_user
         self._rhodecode_user = auth_user
 
         log.info('IP: %s User: %s accessed %s [%s]' % (
             self.ip_addr, auth_user, safe_unicode(get_access_path(environ)),
             _route_name)
         )
 
         user_obj = auth_user.get_instance()
         if user_obj and user_obj.user_data.get('force_password_change'):
             h.flash('You are required to change your password', 'warning',
                     ignore_duplicate=True)
             return self._dispatch_redirect(
                 url('my_account_password'), environ, start_response)
 
         return WSGIController.__call__(self, environ, start_response)
 
 
 def h_filter(s):
     """
     Custom filter for Mako templates. Mako by standard uses `markupsafe.escape`
     we wrap this with additional functionality that converts None to empty
     strings
     """
     if s is None:
         return markupsafe.Markup()
     return markupsafe.escape(s)
 
 
 def add_events_routes(config):
     """
     Adds routing that can be used in events. Because some events are triggered
     outside of pyramid context, we need to bootstrap request with some
     routing registered
     """
     config.add_route(name='home', pattern='/')
 
     config.add_route(name='repo_summary', pattern='/{repo_name}')
     config.add_route(name='repo_summary_explicit', pattern='/{repo_name}/summary')
     config.add_route(name='repo_group_home', pattern='/{repo_group_name}')
 
     config.add_route(name='pullrequest_show',
                      pattern='/{repo_name}/pull-request/{pull_request_id}')
     config.add_route(name='pull_requests_global',
                      pattern='/pull-request/{pull_request_id}')
 
     config.add_route(name='repo_commit',
                      pattern='/{repo_name}/changeset/{commit_id}')
     config.add_route(name='repo_files',
                      pattern='/{repo_name}/files/{commit_id}/{f_path}')
 
 
 def bootstrap_request(**kwargs):
     import pyramid.testing
-    request = pyramid.testing.DummyRequest(**kwargs)
-    request.application_url = kwargs.pop('application_url', 'http://example.com')
-    request.host = kwargs.pop('host', 'example.com:80')
-    request.domain = kwargs.pop('domain', 'example.com')
+
+    class TestRequest(pyramid.testing.DummyRequest):
+        application_url = kwargs.pop('application_url', 'http://example.com')
+        host = kwargs.pop('host', 'example.com:80')
+        domain = kwargs.pop('domain', 'example.com')
+
+    class TestDummySession(pyramid.testing.DummySession):
+        def save(*arg, **kw):
+            pass
+
+    request = TestRequest(**kwargs)
+    request.session = TestDummySession()
 
     config = pyramid.testing.setUp(request=request)
     add_events_routes(config)
+    return request
+

rhodecode/lib/helpers.py

@@ -1,2062 +1,2102 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 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 random
 import hashlib
 import StringIO
 import urllib
 import math
 import logging
 import re
 import urlparse
 import time
 import string
 import hashlib
 from collections import OrderedDict
 
 import pygments
 import itertools
 import fnmatch
 
 from datetime import datetime
 from functools import partial
 from pygments.formatters.html import HtmlFormatter
 from pygments import highlight as code_highlight
 from pygments.lexers import (
     get_lexer_by_name, get_lexer_for_filename, get_lexer_for_mimetype)
 
 from pyramid.threadlocal import get_current_request
 
 from webhelpers.html import literal, HTML, escape
 from webhelpers.html.tools import *
 from webhelpers.html.builder import make_tag
 from webhelpers.html.tags import auto_discovery_link, checkbox, css_classes, \
     end_form, file, form as wh_form, hidden, image, javascript_link, link_to, \
     link_to_if, link_to_unless, ol, required_legend, select, stylesheet_link, \
     submit, text, password, textarea, title, ul, xml_declaration, radio
 from webhelpers.html.tools import auto_link, button_to, highlight, \
     js_obfuscate, mail_to, strip_links, strip_tags, tag_re
-from webhelpers.pylonslib import Flash as _Flash
 from webhelpers.text import chop_at, collapse, convert_accented_entities, \
     convert_misc_entities, lchop, plural, rchop, remove_formatting, \
     replace_whitespace, urlify, truncate, wrap_paragraphs
 from webhelpers.date import time_ago_in_words
 from webhelpers.paginate import Page as _Page
 from webhelpers.html.tags import _set_input_attrs, _set_id_attr, \
     convert_boolean_attrs, NotGiven, _make_safe_id_component
 from webhelpers2.number import format_byte_size
 
 from rhodecode.lib.action_parser import action_parser
 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
 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.config.conf import DATE_FORMAT, DATETIME_FORMAT
 from rhodecode.model.changeset_status import ChangesetStatusModel
 from rhodecode.model.db import Permission, User, Repository
 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 url(*args, **kw):
     from pylons import url as pylons_url
     return pylons_url(*args, **kw)
 
 
 def pylons_url_current(*args, **kw):
     """
     This function overrides pylons.url.current() which returns the current
     path so that it will also work from a pyramid only context. This
     should be removed once port to pyramid is complete.
     """
     from pylons import url as pylons_url
     if not args and not kw:
         request = get_current_request()
         return request.path
     return pylons_url.current(*args, **kw)
 
 url.current = pylons_url_current
 
 
 def url_replace(**qargs):
     """ Returns the current request url while replacing query string args """
 
     request = get_current_request()
     new_args = request.GET.mixed()
     new_args.update(qargs)
     return url('', **new_args)
 
 
 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):
     postfix = '...'
     if len(text) > size:
         return text[:size - len(postfix)] + postfix
     return text
 
 
 def _reset(name, value=None, id=NotGiven, type="reset", **attrs):
     """
     Reset button
     """
     _set_input_attrs(attrs, type, name, value)
     _set_id_attr(attrs, id, name)
     convert_boolean_attrs(attrs, ["disabled"])
     return HTML.input(**attrs)
 
 reset = _reset
 safeid = _make_safe_id_component
 
 
 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()
 
 
 def files_breadcrumbs(repo_name, commit_id, file_path):
     if isinstance(file_path, str):
         file_path = safe_unicode(file_path)
 
     # TODO: johbo: Is this always a url like path, or is this operating
     # system dependent?
     path_segments = file_path.split('/')
 
     repo_name_html = escape(repo_name)
     if len(path_segments) == 1 and path_segments[0] == '':
         url_segments = [repo_name_html]
     else:
         url_segments = [
             link_to(
                 repo_name_html,
                 route_path(
                     'repo_files',
                     repo_name=repo_name,
                     commit_id=commit_id,
                     f_path=''),
                 class_='pjax-link')]
 
     last_cnt = len(path_segments) - 1
     for cnt, segment in enumerate(path_segments):
         if not segment:
             continue
         segment_html = escape(segment)
 
         if cnt != last_cnt:
             url_segments.append(
                 link_to(
                     segment_html,
                     route_path(
                         'repo_files',
                         repo_name=repo_name,
                         commit_id=commit_id,
                         f_path='/'.join(path_segments[:cnt + 1])),
                     class_='pjax-link'))
         else:
             url_segments.append(segment_html)
 
     return literal('/'.join(url_segments))
 
 
 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 extract_phrases(text_query):
     """
     Extracts phrases from search term string making sure phrases
     contained in double quotes are kept together - and discarding empty values
     or fully whitespace values eg.
 
     'some   text "a phrase" more' => ['some', 'text', 'a phrase', 'more']
 
     """
 
     in_phrase = False
     buf = ''
     phrases = []
     for char in text_query:
         if in_phrase:
             if char == '"': # end phrase
                 phrases.append(buf)
                 buf = ''
                 in_phrase = False
                 continue
             else:
                 buf += char
                 continue
         else:
             if char == '"': # start phrase
                 in_phrase = True
                 phrases.append(buf)
                 buf = ''
                 continue
             elif char == ' ':
                 phrases.append(buf)
                 buf = ''
                 continue
             else:
                 buf += char
 
     phrases.append(buf)
     phrases = [phrase.strip() for phrase in phrases if phrase.strip()]
     return phrases
 
 
 def get_matching_offsets(text, phrases):
     """
     Returns a list of string offsets in `text` that the list of `terms` match
 
     >>> get_matching_offsets('some text here', ['some', 'here'])
     [(0, 4), (10, 14)]
 
     """
     offsets = []
     for phrase in phrases:
         for match in re.finditer(phrase, text):
             offsets.append((match.start(), match.end()))
 
     return offsets
 
 
 def normalize_text_for_matching(x):
     """
     Replaces all non alnum characters to spaces and lower cases the string,
     useful for comparing two text strings without punctuation
     """
     return re.sub(r'[^\w]', ' ', x.lower())
 
 
 def get_matching_line_offsets(lines, terms):
     """ Return a set of `lines` indices (starting from 1) matching a
     text search query, along with `context` lines above/below matching lines
 
     :param lines: list of strings representing lines
     :param terms: search term string to match in lines eg. 'some text'
     :param context: number of lines above/below a matching line to add to result
     :param max_lines: cut off for lines of interest
      eg.
 
     text = '''
     words words words
     words words words
     some text some
     words words words
     words words words
     text here what
     '''
     get_matching_line_offsets(text, 'text', context=1)
     {3: [(5, 9)], 6: [(0, 4)]]
 
     """
     matching_lines = {}
     phrases = [normalize_text_for_matching(phrase)
                for phrase in extract_phrases(terms)]
 
     for line_index, line in enumerate(lines, start=1):
         match_offsets = get_matching_offsets(
             normalize_text_for_matching(line), phrases)
         if match_offsets:
             matching_lines[line_index] = match_offsets
 
     return matching_lines
 
 
 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 xrange(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):
         self.category = category
         self.message = message
 
     def __str__(self):
         return self.message
 
     __unicode__ = __str__
 
     def __html__(self):
         return escape(safe_unicode(self.message))
 
 
-class Flash(_Flash):
+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.
 
-    def pop_messages(self, request=None):
-        """Return all accumulated messages and delete them from the 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 request:
+        if not session:
+            if not request:
+                request = get_current_request()
             session = request.session
-        else:
-            from pylons import 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):
                 messages.append(_Message(cat, msg))
         # 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, request=None):
+    def json_alerts(self, session=None, request=None):
         payloads = []
-        messages = flash.pop_messages(request=request)
+        messages = flash.pop_messages(session=session, request=request)
         if messages:
             for message in messages:
                 subdata = {}
                 if hasattr(message.message, 'rsplit'):
                     flash_data = message.message.rsplit('|DELIM|', 1)
                     org_message = flash_data[0]
                     if len(flash_data) > 1:
                         subdata = json.loads(flash_data[1])
                 else:
                     org_message = message.message
                 payloads.append({
                     'message': {
                         'message': u'{}'.format(org_message),
                         'level': message.category,
                         'force': True,
                         'subdata': subdata
                     }
                 })
         return json.dumps(payloads)
 
+    def __call__(self, message, category=None, ignore_duplicate=False,
+                 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 credentials_filter, age as _age
 from rhodecode.model.db import User, ChangesetStatus
 
 age = _age
 capitalize = lambda x: x.capitalize()
 email = author_email
 short_id = lambda x: x[:12]
 hide_credentials = lambda x: ''.join(credentials_filter(x))
 
 
 def age_component(datetime_iso, value=None, time_is_local=False):
     title = value or format_date(datetime_iso)
     tzinfo = '+00:00'
 
     # detect if we have a timezone info, otherwise, add it
     if isinstance(datetime_iso, datetime) and not datetime_iso.tzinfo:
         if time_is_local:
             tzinfo = time.strftime("+%H:%M",
                 time.gmtime(
                     (datetime.now() - datetime.utcnow()).seconds + 1
                 )
             )
 
     return literal(
         '<time class="timeago tooltip" '
         'title="{1}{2}" datetime="{0}{2}">{1}</time>'.format(
             datetime_iso, title, tzinfo))
 
 
 def _shorten_commit_id(commit_id):
     from rhodecode import CONFIG
     def_len = safe_int(CONFIG.get('rhodecode_show_sha_length', 12))
     return commit_id[:def_len]
 
 
 def show_id(commit):
     """
     Configurable function that shows ID
     by default it's r123:fffeeefffeee
 
     :param commit: commit instance
     """
     from rhodecode import CONFIG
     show_idx = str2bool(CONFIG.get('rhodecode_show_revision_number', True))
 
     raw_id = _shorten_commit_id(commit.raw_id)
     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)
     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 autho 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:
         return link_to(
             escape(display_person),
             route_path('user_profile', username=user.username),
             **kwargs)
     else:
         return escape(display_person)
 
 
 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):
     _render = request.get_partial_renderer('base/base.mako')
     return _render('gravatar_with_user', author, show_disabled=show_disabled)
 
 
 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\ \.\-\_]+)\]\((.*?)\)\]'),
             '<div class="metatag" tag="url"> <a href="\\2">\\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
     """
     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):
     """
     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")
     else:  # not true as bool
         return HTML.tag('i', class_="icon-false")
 
 
 #==============================================================================
 # PERMS
 #==============================================================================
 from rhodecode.lib.auth import HasPermissionAny, HasPermissionAll, \
 HasRepoPermissionAny, HasRepoPermissionAll, HasRepoGroupPermissionAll, \
 HasRepoGroupPermissionAny, HasRepoPermissionAnyApi, get_csrf_token, \
 csrf_token_key
 
 
 #==============================================================================
 # 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
         """
 
         font_family = ','.join([
             'proximanovaregular',
             'Proxima Nova Regular',
             'Proxima Nova',
             'Arial',
             'Lucida Grande',
             'sans-serif'
             ])
         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/1.85,  # 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" % img_data.encode('base64')
 
 
 def initials_gravatar(email_address, first_name, last_name, size=30):
     svg_type = None
     if email_address == User.DEFAULT_USER_EMAIL:
         svg_type = 'default_user'
     klass = InitialsGravatar(email_address, first_name, last_name, size)
     return klass.generate_svg(svg_type=svg_type)
 
 
 def gravatar_url(email_address, size=30, request=None):
     request = get_current_request()
     if request and hasattr(request, 'call_context'):
         _use_gravatar = request.call_context.visual.use_gravatar
         _gravatar_url = request.call_context.visual.gravatar_url
     else:
         # doh, we need to re-import those to mock it later
         from pylons import tmpl_context as c
 
         _use_gravatar = c.visual.use_gravatar
         _gravatar_url = c.visual.gravatar_url
 
     _gravatar_url = _gravatar_url or User.DEFAULT_GRAVATAR_URL
 
     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(User.DEFAULT_USER_EMAIL, '', '', size=size)
 
     if _use_gravatar:
         # TODO: Disuse pyramid thread locals. Think about another solution to
         # get the host and schema here.
         request = get_current_request()
         tmpl = safe_str(_gravatar_url)
         tmpl = 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))
         return tmpl
     else:
         return initials_gravatar(email_address, '', '', size=size)
 
 
 class Page(_Page):
     """
     Custom pager to match rendering style with paginator
     """
 
     def _get_pos(self, cur_page, max_page, items):
         edge = (items / 2) + 1
         if (cur_page <= edge):
             radius = max(items / 2, items - cur_page)
         elif (max_page - cur_page) < edge:
             radius = (items - 1) - (max_page - cur_page)
         else:
             radius = items / 2
 
         left = max(1, (cur_page - (radius)))
         right = min(max_page, cur_page + (radius))
         return left, cur_page, right
 
     def _range(self, regexp_match):
         """
         Return range of linked pages (e.g. '1 2 [3] 4 5 6 7 8').
 
         Arguments:
 
         regexp_match
             A "re" (regular expressions) match object containing the
             radius of linked pages around the current page in
             regexp_match.group(1) as a string
 
         This function is supposed to be called as a callable in
         re.sub.
 
         """
         radius = int(regexp_match.group(1))
 
         # Compute the first and last page number within the radius
         # e.g. '1 .. 5 6 [7] 8 9 .. 12'
         # -> leftmost_page  = 5
         # -> rightmost_page = 9
         leftmost_page, _cur, rightmost_page = self._get_pos(self.page,
                                                             self.last_page,
                                                             (radius * 2) + 1)
         nav_items = []
 
         # Create a link to the first page (unless we are on the first page
         # or there would be no need to insert '..' spacers)
         if self.page != self.first_page and self.first_page < leftmost_page:
             nav_items.append(self._pagerlink(self.first_page, self.first_page))
 
         # Insert dots if there are pages between the first page
         # and the currently displayed page range
         if leftmost_page - self.first_page > 1:
             # Wrap in a SPAN tag if nolink_attr is set
             text = '..'
             if self.dotdot_attr:
                 text = HTML.span(c=text, **self.dotdot_attr)
             nav_items.append(text)
 
         for thispage in xrange(leftmost_page, rightmost_page + 1):
             # Hilight the current page number and do not use a link
             if thispage == self.page:
                 text = '%s' % (thispage,)
                 # Wrap in a SPAN tag if nolink_attr is set
                 if self.curpage_attr:
                     text = HTML.span(c=text, **self.curpage_attr)
                 nav_items.append(text)
             # Otherwise create just a link to that page
             else:
                 text = '%s' % (thispage,)
                 nav_items.append(self._pagerlink(thispage, text))
 
         # Insert dots if there are pages between the displayed
         # page numbers and the end of the page range
         if self.last_page - rightmost_page > 1:
             text = '..'
             # Wrap in a SPAN tag if nolink_attr is set
             if self.dotdot_attr:
                 text = HTML.span(c=text, **self.dotdot_attr)
             nav_items.append(text)
 
         # Create a link to the very last page (unless we are on the last
         # page or there would be no need to insert '..' spacers)
         if self.page != self.last_page and rightmost_page < self.last_page:
             nav_items.append(self._pagerlink(self.last_page, self.last_page))
 
         ## prerender links
         #_page_link = url.current()
         #nav_items.append(literal('<link rel="prerender" href="%s?page=%s">' % (_page_link, str(int(self.page)+1))))
         #nav_items.append(literal('<link rel="prefetch" href="%s?page=%s">' % (_page_link, str(int(self.page)+1))))
         return self.separator.join(nav_items)
 
     def pager(self, format='~2~', page_param='page', partial_param='partial',
         show_if_single_page=False, separator=' ', onclick=None,
         symbol_first='<<', symbol_last='>>',
         symbol_previous='<', symbol_next='>',
         link_attr={'class': 'pager_link', 'rel': 'prerender'},
         curpage_attr={'class': 'pager_curpage'},
         dotdot_attr={'class': 'pager_dotdot'}, **kwargs):
 
         self.curpage_attr = curpage_attr
         self.separator = separator
         self.pager_kwargs = kwargs
         self.page_param = page_param
         self.partial_param = partial_param
         self.onclick = onclick
         self.link_attr = link_attr
         self.dotdot_attr = dotdot_attr
 
         # Don't show navigator if there is no more than one page
         if self.page_count == 0 or (self.page_count == 1 and not show_if_single_page):
             return ''
 
         from string import Template
         # Replace ~...~ in token format by range of pages
         result = re.sub(r'~(\d+)~', self._range, format)
 
         # Interpolate '%' variables
         result = Template(result).safe_substitute({
             'first_page': self.first_page,
             'last_page': self.last_page,
             'page': self.page,
             'page_count': self.page_count,
             'items_per_page': self.items_per_page,
             'first_item': self.first_item,
             'last_item': self.last_item,
             'item_count': self.item_count,
             'link_first': self.page > self.first_page and \
                     self._pagerlink(self.first_page, symbol_first) or '',
             'link_last': self.page < self.last_page and \
                     self._pagerlink(self.last_page, symbol_last) or '',
             'link_previous': self.previous_page and \
                     self._pagerlink(self.previous_page, symbol_previous) \
                     or HTML.span(symbol_previous, class_="pg-previous disabled"),
             'link_next': self.next_page and \
                     self._pagerlink(self.next_page, symbol_next) \
                     or HTML.span(symbol_next, class_="pg-next disabled")
         })
 
         return literal(result)
 
 
 #==============================================================================
 # REPO PAGER, PAGER FOR REPOSITORY
 #==============================================================================
 class RepoPage(Page):
 
     def __init__(self, collection, page=1, items_per_page=20,
                  item_count=None, url=None, **kwargs):
 
         """Create a "RepoPage" instance. special pager for paging
         repository
         """
         self._url_generator = url
 
         # Safe the kwargs class-wide so they can be used in the pager() method
         self.kwargs = kwargs
 
         # Save a reference to the collection
         self.original_collection = collection
 
         self.collection = collection
 
         # The self.page is the number of the current page.
         # The first page has the number 1!
         try:
             self.page = int(page)  # make it int() if we get it as a string
         except (ValueError, TypeError):
             self.page = 1
 
         self.items_per_page = items_per_page
 
         # Unless the user tells us how many items the collections has
         # we calculate that ourselves.
         if item_count is not None:
             self.item_count = item_count
         else:
             self.item_count = len(self.collection)
 
         # Compute the number of the first and last available page
         if self.item_count > 0:
             self.first_page = 1
             self.page_count = int(math.ceil(float(self.item_count) /
                                             self.items_per_page))
             self.last_page = self.first_page + self.page_count - 1
 
             # Make sure that the requested page number is the range of
             # valid pages
             if self.page > self.last_page:
                 self.page = self.last_page
             elif self.page < self.first_page:
                 self.page = self.first_page
 
             # Note: the number of items on this page can be less than
             #       items_per_page if the last page is not full
             self.first_item = max(0, (self.item_count) - (self.page *
                                                           items_per_page))
             self.last_item = ((self.item_count - 1) - items_per_page *
                               (self.page - 1))
 
             self.items = list(self.collection[self.first_item:self.last_item + 1])
 
             # Links to previous and next page
             if self.page > self.first_page:
                 self.previous_page = self.page - 1
             else:
                 self.previous_page = None
 
             if self.page < self.last_page:
                 self.next_page = self.page + 1
             else:
                 self.next_page = None
 
         # No items available
         else:
             self.first_page = None
             self.page_count = 0
             self.last_page = None
             self.first_item = None
             self.last_item = None
             self.previous_page = None
             self.next_page = None
             self.items = []
 
         # This is a subclass of the 'list' type. Initialise the list now.
         list.__init__(self, reversed(self.items))
 
 
 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))
         for group in repo.groups_with_parents
     ] + [
         link_to(repo.just_name, route_path('repo_summary', repo_name=repo.repo_name))
     ]
 
     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):
     """
     Extrac urls from text and make html links out of them
 
     :param text_:
     """
 
     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]
         return '<a href="%(url)s">%(url)s</a>' % ({'url': url_full})
     _newtext = url_pat.sub(url_func, text_)
     if safe:
         return literal(_newtext)
     return _newtext
 
 
 def urlify_commits(text_, repository):
     """
     Extract commit ids from text and make link from them
 
     :param text_:
     :param repository: 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="%(cls)s" href="%(url)s">'
             '%(commit_id)s</a>%(suf)s'
         )
         return tmpl % {
             'pref': pref,
             'cls': 'revision-link',
             'url': route_url('repo_commit', repo_name=repository,
                              commit_id=commit_id),
             'commit_id': commit_id,
             'suf': suf
         }
 
     newtext = URL_PAT.sub(url_func, text_)
 
     return newtext
 
 
 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="%(cls)s" href="%(url)s">'
             '%(issue-prefix)s%(id-repr)s'
             '</a>')
     elif link_format == 'rst':
         tmpl = '`%(issue-prefix)s%(id-repr)s <%(url)s>`_'
     elif link_format == 'markdown':
         tmpl = '[%(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
     }
     # named regex variables
     named_vars.update(match_obj.groupdict())
     _url = string.Template(entry['url']).safe_substitute(**named_vars)
 
     data = {
         'pref': pref,
         'cls': 'issue-tracker-link',
         'url': _url,
         'id-repr': issue_id,
         'issue-prefix': entry['pref'],
         'serv': entry['url'],
     }
     if return_raw_data:
         return {
             'id': issue_id,
             'url': _url
         }
     return tmpl % data
 
 
 def process_patterns(text_string, repo_name, link_format='html'):
     allowed_formats = ['html', 'rst', 'markdown']
     if link_format not in allowed_formats:
         raise ValueError('Link format can be only one of:{} got {}'.format(
                          allowed_formats, link_format))
 
     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)
 
     issues_data = []
     newtext = text_string
 
     for uid, entry in active_entries.items():
         log.debug('found issue tracker entry with uid %s' % (uid,))
 
         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']))
 
         try:
             pattern = re.compile(r'%s' % entry['pat'])
         except re.error:
             log.exception(
                 'issue tracker pattern: `%s` failed to compile',
                 entry['pat'])
             continue
 
         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)
 
         newtext = pattern.sub(url_func, newtext)
         log.debug('processed prefix:uid `%s`' % (uid,))
 
     return newtext, issues_data
 
 
 def urlify_commit_message(commit_text, repository=None):
     """
     Parses given text message and makes proper links.
     issues are linked to given issue-server, and rest is a commit link
 
     :param commit_text:
     :param repository:
     """
     from pylons import url  # doh, we need to re-import url to mock it later
 
     def escaper(string):
         return string.replace('<', '&lt;').replace('>', '&gt;')
 
     newtext = escaper(commit_text)
 
     # extract http/https links and make them real urls
     newtext = urlify_text(newtext, safe=False)
 
     # urlify commits - extract commit ids and make link out of them, if we have
     # the scope of repository present.
     if repository:
         newtext = urlify_commits(newtext, repository)
 
     # process issue tracker patterns
     newtext, issues = process_patterns(newtext, repository or '')
 
     return literal(newtext)
 
 
 def render_binary(repo_name, file_obj):
     """
     Choose how to render a binary file
     """
     filename = file_obj.name
 
     # images
     for ext in ['*.png', '*.jpg', '*.ico', '*.gif']:
         if fnmatch.fnmatch(filename, pat=ext):
             alt = filename
             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="{}" src="{}">'.format(alt, 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):
 
     def maybe_convert_relative_links(html_source):
         if relative_urls:
             return relative_links(html_source, relative_urls)
         return html_source
 
     if renderer == 'rst':
         if repo_name:
             # process patterns on comments if we pass in repo name
             source, issues = process_patterns(
                 source, repo_name, link_format='rst')
 
         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 = process_patterns(
                 source, repo_name, link_format='markdown')
 
         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
 
     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' +
         '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'
     )
 
 
 def search_filter_help(searcher, request):
     _ = request.translate
 
     terms = ''
     return _(
         'Example filter terms for `{searcher}` search:\n' +
         '{terms}\n' +
         'Generate wildcards using \'*\' character:\n' +
         '     "repo_name:vcs*" - search everything starting with \'vcs\'\n' +
         '     "repo_name:*vcs*" - search for repository containing \'vcs\'\n' +
         '\n' +
         'Optional AND / OR operators in queries\n' +
         '     "repo_name:vcs OR repo_name:test"\n' +
         '     "owner:test AND repo_name:test*"\n' +
         'More: {search_doc}'
     ).format(searcher=searcher.name,
              terms=terms, search_doc=searcher.query_lang_doc)
 
 
 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 wh_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.
 
     """
     from webhelpers.pylonslib.secure_form import insecure_form
 
     session = None
 
     # TODO(marcink): after pyramid migration require request variable ALWAYS
     if 'request' in attrs:
         session = attrs['request'].session
         del attrs['request']
 
     form = insecure_form(form_url, method, multipart, **attrs)
     token = literal(
         '<input type="hidden" id="{}" name="{}" value="{}">'.format(
         csrf_token_key, 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',
                       dropdownCssClass: 'drop-menu-dropdown',
                       dropdownAutoWidth: true%s
                   });
             });
         </script>
     """
     filter_option = """,
                         minimumResultsForSearch: -1
     """
     input_id = attrs.get('id') or name
     filter_enabled = "" if enable_filter else filter_option
     select_script = literal(select2 % (input_id, 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. 
     It is used to generate URLs from within pylons views or templates. 
     This will be removed when pyramid migration if finished.
     """
     req = get_current_request()
     return req.route_url(*args, **kwargs)
 
 
 def route_path(*args, **kwargs):
     """
     Wrapper around pyramids `route_path` function. It is used to generate
     URLs from within pylons views or templates. This will be removed when
     pyramid migration if finished.
     """
     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 static_url(*args, **kwds):
     """
     Wrapper around pyramids `route_path` function. It is used to generate
     URLs from within pylons views or templates. This will be removed when
     pyramid migration if finished.
     """
     req = get_current_request()
     return req.static_url(*args, **kwds)
 
 
 def resource_path(*args, **kwds):
     """
     Wrapper around pyramids `route_path` function. It is used to generate
     URLs from within pylons views or templates. This will be removed when
     pyramid migration if finished.
     """
     req = get_current_request()
     return req.resource_path(*args, **kwds)
 
 
 def api_call_example(method, args):
     """
     Generates an API call example via CURL
     """
     args_json = json.dumps(OrderedDict([
         ('id', 1),
         ('auth_token', 'SECRET'),
         ('method', method),
         ('args', args)
     ]))
     return literal(
         "curl {api_url} -X POST -H 'content-type:text/plain' --data-binary '{data}'"
         "<br/><br/>SECRET can be found in <a href=\"{token_url}\">auth-tokens</a> page, "
         "and needs to be of `api calls` role."
         .format(
             api_url=route_url('apiv2'),
             token_url=route_url('my_account_auth_tokens'),
             data=args_json))
 
 
 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)

rhodecode/templates/base/root.mako

@@ -1,183 +1,183 b''
 ## -*- coding: utf-8 -*-
 <!DOCTYPE html>
 
 <%
 c.template_context['repo_name'] = getattr(c, 'repo_name', '')
 
 if hasattr(c, 'rhodecode_db_repo'):
     c.template_context['repo_type'] = c.rhodecode_db_repo.repo_type
     c.template_context['repo_landing_commit'] = c.rhodecode_db_repo.landing_rev[1]
 
 if getattr(c, 'rhodecode_user', None) and c.rhodecode_user.user_id:
     c.template_context['rhodecode_user']['username'] = c.rhodecode_user.username
     c.template_context['rhodecode_user']['email'] = c.rhodecode_user.email
     c.template_context['rhodecode_user']['notification_status'] = c.rhodecode_user.get_instance().user_data.get('notification_status', True)
     c.template_context['rhodecode_user']['first_name'] = c.rhodecode_user.first_name
     c.template_context['rhodecode_user']['last_name'] = c.rhodecode_user.last_name
 
 c.template_context['visual']['default_renderer'] = h.get_visual_attr(c, 'default_renderer')
 c.template_context['default_user'] = {
     'username': h.DEFAULT_USER,
     'user_id': 1
 }
 
 %>
 <html xmlns="http://www.w3.org/1999/xhtml">
     <head>
         <script src="${h.asset('js/vendors/webcomponentsjs/webcomponents-lite.min.js', ver=c.rhodecode_version_hash)}"></script>
         <link rel="import" href="${h.asset('js/rhodecode-components.html', ver=c.rhodecode_version_hash)}">
         <title>${self.title()}</title>
         <meta http-equiv="Content-Type" content="text/html;charset=utf-8" />
 
         % if 'safari' in (request.user_agent or '').lower():
             <meta name="referrer" content="origin">
         % else:
             <meta name="referrer" content="origin-when-cross-origin">
         % endif
 
         <%def name="robots()">
             <meta name="robots" content="index, nofollow"/>
         </%def>
         ${self.robots()}
         <link rel="icon" href="${h.asset('images/favicon.ico', ver=c.rhodecode_version_hash)}" sizes="16x16 32x32" type="image/png" />
 
         ## CSS definitions
         <%def name="css()">
             <link rel="stylesheet" type="text/css" href="${h.asset('css/style.css', ver=c.rhodecode_version_hash)}" media="screen"/>
             <!--[if lt IE 9]>
                 <link rel="stylesheet" type="text/css" href="${h.asset('css/ie.css', ver=c.rhodecode_version_hash)}" media="screen"/>
             <![endif]-->
             ## EXTRA FOR CSS
             ${self.css_extra()}
         </%def>
         ## CSS EXTRA - optionally inject some extra CSS stuff needed for specific websites
         <%def name="css_extra()">
         </%def>
 
         ${self.css()}
 
         ## JAVASCRIPT
         <%def name="js()">
             <script>
                 // setup Polymer options
                 window.Polymer = {lazyRegister: true, dom: 'shadow'};
 
                 // Load webcomponentsjs polyfill if browser does not support native Web Components
                 (function() {
                     'use strict';
                     var onload = function() {
                         // For native Imports, manually fire WebComponentsReady so user code
                         // can use the same code path for native and polyfill'd imports.
                         if (!window.HTMLImports) {
                             document.dispatchEvent(
                                     new CustomEvent('WebComponentsReady', {bubbles: true})
                             );
                         }
                     };
                     var webComponentsSupported = (
                             'registerElement' in document
                             && 'import' in document.createElement('link')
                             && 'content' in document.createElement('template')
                     );
                     if (!webComponentsSupported) {
                     } else {
                         onload();
                     }
                 })();
             </script>
 
             <script src="${h.asset('js/rhodecode/i18n/%s.js' % c.language, ver=c.rhodecode_version_hash)}"></script>
             <script type="text/javascript">
             // register templateContext to pass template variables to JS
             var templateContext = ${h.json.dumps(c.template_context)|n};
 
             var APPLICATION_URL = "${h.route_path('home').rstrip('/')}";
             var ASSET_URL = "${h.asset('')}";
             var DEFAULT_RENDERER = "${h.get_visual_attr(c, 'default_renderer')}";
             var CSRF_TOKEN = "${getattr(c, 'csrf_token', '')}";
 
             var APPENLIGHT = {
               enabled: ${'true' if getattr(c, 'appenlight_enabled', False) else 'false'},
               key: '${getattr(c, "appenlight_api_public_key", "")}',
               % if getattr(c, 'appenlight_server_url', None):
                   serverUrl: '${getattr(c, "appenlight_server_url", "")}',
               % endif
               requestInfo: {
               % if getattr(c, 'rhodecode_user', None):
                   ip: '${c.rhodecode_user.ip_addr}',
                   username: '${c.rhodecode_user.username}'
               % endif
               },
               tags: {
                   rhodecode_version: '${c.rhodecode_version}',
                   rhodecode_edition: '${c.rhodecode_edition}'
               }
             };
 
             </script>
             <%include file="/base/plugins_base.mako"/>
             <!--[if lt IE 9]>
             <script language="javascript" type="text/javascript" src="${h.asset('js/excanvas.min.js')}"></script>
             <![endif]-->
             <script language="javascript" type="text/javascript" src="${h.asset('js/rhodecode/routes.js', ver=c.rhodecode_version_hash)}"></script>
-            <script> var alertMessagePayloads = ${h.flash.json_alerts(request)|n}; </script>
+            <script> var alertMessagePayloads = ${h.flash.json_alerts(request=request)|n}; </script>
             ## avoide escaping the %N
             <script language="javascript" type="text/javascript" src="${h.asset('js/rhodecode-components.js', ver=c.rhodecode_version_hash)}"></script>
             <script>CodeMirror.modeURL = "${h.asset('') + 'js/mode/%N/%N.js?ver='+c.rhodecode_version_hash}";</script>
 
 
             ## JAVASCRIPT EXTRA - optionally inject some extra JS for specificed templates
             ${self.js_extra()}
 
             <script type="text/javascript">
             Rhodecode = (function() {
                 function _Rhodecode() {
                   this.comments = new CommentsController();
                 }
                 return new _Rhodecode();
             })();
 
             $(document).ready(function(){
               show_more_event();
               timeagoActivate();
               clipboardActivate();
             })
             </script>
 
         </%def>
 
         ## JAVASCRIPT EXTRA - optionally inject some extra JS for specificed templates
         <%def name="js_extra()"></%def>
         ${self.js()}
 
         <%def name="head_extra()"></%def>
         ${self.head_extra()}
         ## extra stuff
         %if c.pre_code:
             ${c.pre_code|n}
         %endif
     </head>
     <body id="body">
         <noscript>
             <div class="noscript-error">
                 ${_('Please enable JavaScript to use RhodeCode Enterprise')}
             </div>
         </noscript>
      ## IE hacks
       <!--[if IE 7]>
       <script>$(document.body).addClass('ie7')</script>
       <![endif]-->
       <!--[if IE 8]>
       <script>$(document.body).addClass('ie8')</script>
       <![endif]-->
       <!--[if IE 9]>
       <script>$(document.body).addClass('ie9')</script>
       <![endif]-->
 
       ${next.body()}
       %if c.post_code:
         ${c.post_code|n}
       %endif
     <rhodecode-app></rhodecode-app>
     </body>
 </html>

rhodecode/tests/__init__.py

@@ -1,265 +1,248 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 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 os
 import time
 import logging
 import datetime
 import hashlib
 import tempfile
 from os.path import join as jn
 
 from tempfile import _RandomNameSequence
 
-from paste.deploy import loadapp
-from paste.script.appinstall import SetupCommand
+from pylons import url
 
-import pylons
-import pylons.test
-from pylons import config, url
-from pylons.i18n.translation import _get_translator
-from pylons.util import ContextObj
-
-from routes.util import URLGenerator
 from nose.plugins.skip import SkipTest
 import pytest
 
-from rhodecode import is_windows
-from rhodecode.config.routing import ADMIN_PREFIX
-from rhodecode.model.meta import Session
 from rhodecode.model.db import User
 from rhodecode.lib import auth
 from rhodecode.lib import helpers as h
 from rhodecode.lib.helpers import flash, link_to
-from rhodecode.lib.utils2 import safe_unicode, safe_str
+from rhodecode.lib.utils2 import safe_str
 
 
 log = logging.getLogger(__name__)
 
 __all__ = [
     'get_new_dir', 'TestController', 'SkipTest',
     'url', 'link_to', 'ldap_lib_installed', 'clear_all_caches',
     'assert_session_flash', 'login_user', 'no_newline_id_generator',
     'TESTS_TMP_PATH', 'HG_REPO', 'GIT_REPO', 'SVN_REPO',
     'NEW_HG_REPO', 'NEW_GIT_REPO',
     'HG_FORK', 'GIT_FORK', 'TEST_USER_ADMIN_LOGIN', 'TEST_USER_ADMIN_PASS',
     'TEST_USER_REGULAR_LOGIN', 'TEST_USER_REGULAR_PASS',
     'TEST_USER_REGULAR_EMAIL', 'TEST_USER_REGULAR2_LOGIN',
     'TEST_USER_REGULAR2_PASS', 'TEST_USER_REGULAR2_EMAIL', 'TEST_HG_REPO',
     'TEST_HG_REPO_CLONE', 'TEST_HG_REPO_PULL', 'TEST_GIT_REPO',
     'TEST_GIT_REPO_CLONE', 'TEST_GIT_REPO_PULL', 'SCM_TESTS',
 ]
 
 # Invoke websetup with the current config file
 # SetupCommand('setup-app').run([config_file])
 
 # SOME GLOBALS FOR TESTS
 TEST_DIR = tempfile.gettempdir()
 
 TESTS_TMP_PATH = jn(TEST_DIR, 'rc_test_%s' % _RandomNameSequence().next())
 TEST_USER_ADMIN_LOGIN = 'test_admin'
 TEST_USER_ADMIN_PASS = 'test12'
 TEST_USER_ADMIN_EMAIL = '[e-mail]'
 
 TEST_USER_REGULAR_LOGIN = 'test_regular'
 TEST_USER_REGULAR_PASS = 'test12'
 TEST_USER_REGULAR_EMAIL = '[e-mail]'
 
 TEST_USER_REGULAR2_LOGIN = 'test_regular2'
 TEST_USER_REGULAR2_PASS = 'test12'
 TEST_USER_REGULAR2_EMAIL = '[e-mail]'
 
 HG_REPO = 'vcs_test_hg'
 GIT_REPO = 'vcs_test_git'
 SVN_REPO = 'vcs_test_svn'
 
 NEW_HG_REPO = 'vcs_test_hg_new'
 NEW_GIT_REPO = 'vcs_test_git_new'
 
 HG_FORK = 'vcs_test_hg_fork'
 GIT_FORK = 'vcs_test_git_fork'
 
 ## VCS
 SCM_TESTS = ['hg', 'git']
 uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))
 
 TEST_GIT_REPO = jn(TESTS_TMP_PATH, GIT_REPO)
 TEST_GIT_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)
 TEST_GIT_REPO_PULL = jn(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)
 
 TEST_HG_REPO = jn(TESTS_TMP_PATH, HG_REPO)
 TEST_HG_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)
 TEST_HG_REPO_PULL = jn(TESTS_TMP_PATH, 'vcshgpull%s' % uniq_suffix)
 
 TEST_REPO_PREFIX = 'vcs-test'
 
 
 # skip ldap tests if LDAP lib is not installed
 ldap_lib_installed = False
 try:
     import ldap
     ldap_lib_installed = True
 except ImportError:
+    ldap = None
     # means that python-ldap is not installed
     pass
 
 
 def clear_all_caches():
     from beaker.cache import cache_managers
     for _cache in cache_managers.values():
         _cache.clear()
 
 
 def get_new_dir(title):
     """
     Returns always new directory path.
     """
     from rhodecode.tests.vcs.utils import get_normalized_path
     name_parts = [TEST_REPO_PREFIX]
     if title:
         name_parts.append(title)
     hex_str = hashlib.sha1('%s %s' % (os.getpid(), time.time())).hexdigest()
     name_parts.append(hex_str)
     name = '-'.join(name_parts)
     path = os.path.join(TEST_DIR, name)
     return get_normalized_path(path)
 
 
 @pytest.mark.usefixtures('app', 'index_location')
 class TestController(object):
 
     maxDiff = None
 
     def log_user(self, username=TEST_USER_ADMIN_LOGIN,
                  password=TEST_USER_ADMIN_PASS):
         self._logged_username = username
         self._session = login_user_session(self.app, username, password)
         self.csrf_token = auth.get_csrf_token(self._session)
 
         return self._session['rhodecode_user']
 
     def logout_user(self):
         logout_user_session(self.app, auth.get_csrf_token(self._session))
         self.csrf_token = None
         self._logged_username = None
         self._session = None
 
     def _get_logged_user(self):
         return User.get_by_username(self._logged_username)
 
 
 def login_user_session(
         app, username=TEST_USER_ADMIN_LOGIN, password=TEST_USER_ADMIN_PASS):
 
     response = app.post(
         h.route_path('login'),
         {'username': username, 'password': password})
     if 'invalid user name' in response.body:
         pytest.fail('could not login using %s %s' % (username, password))
 
     assert response.status == '302 Found'
     response = response.follow()
     assert response.status == '200 OK'
 
     session = response.get_session_from_response()
     assert 'rhodecode_user' in session
     rc_user = session['rhodecode_user']
     assert rc_user.get('username') == username
     assert rc_user.get('is_authenticated')
 
     return session
 
 
 def logout_user_session(app, csrf_token):
     app.post(h.route_path('logout'), {'csrf_token': csrf_token}, status=302)
 
 
 def login_user(app, username=TEST_USER_ADMIN_LOGIN,
                password=TEST_USER_ADMIN_PASS):
     return login_user_session(app, username, password)['rhodecode_user']
 
 
-def assert_session_flash(response=None, msg=None, category=None, no_=None):
+def assert_session_flash(response, msg=None, category=None, no_=None):
     """
     Assert on a flash message in the current session.
 
-    :param msg: Required. The expected message. Will be evaluated if a
+    :param response: Response from give calll, it will contain flash
+        messages or bound session with them.
+    :param msg: The expected message. Will be evaluated if a
         :class:`LazyString` is passed in.
-    :param response: Optional. For functional testing, pass in the response
-        object. Otherwise don't pass in any value.
     :param category: Optional. If passed, the message category will be
         checked as well.
-    :param no_: Optional. If passed, the message will be checked to NOT be in the
-        flash session
+    :param no_: Optional. If passed, the message will be checked to NOT
+        be in the flash session
     """
     if msg is None and no_ is None:
         raise ValueError("Parameter msg or no_ is required.")
 
     if msg and no_:
         raise ValueError("Please specify either msg or no_, but not both")
 
-    messages = flash.pop_messages()
+    session = response.get_session_from_response()
+    messages = flash.pop_messages(session=session)
     msg = _eval_if_lazy(msg)
 
     assert messages, 'unable to find message `%s` in empty flash list' % msg
     message = messages[0]
 
     message_text = _eval_if_lazy(message.message) or ''
 
     if no_:
         if no_ in message_text:
             msg = u'msg `%s` found in session flash.' % (no_,)
             pytest.fail(safe_str(msg))
     else:
         if msg not in message_text:
             fail_msg = u'msg `%s` not found in session ' \
                        u'flash: got `%s` (type:%s) instead' % (
                 msg, message_text, type(message_text))
 
             pytest.fail(safe_str(fail_msg))
         if category:
             assert category == message.category
 
 
 def _eval_if_lazy(value):
     return value.eval() if hasattr(value, 'eval') else value
 
 
-def assert_session_flash_is_empty(response):
-    assert 'flash' in response.session, 'Response session has no flash key'
-
-    msg = 'flash messages are present in session:%s' % \
-          response.session['flash'][0]
-    pytest.fail(safe_str(msg))
-
-
 def no_newline_id_generator(test_name):
     """
     Generates a test name without spaces or newlines characters. Used for
     nicer output of progress of test
     """
     org_name = test_name
     test_name = test_name\
         .replace('\n', '_N') \
         .replace('\r', '_N') \
         .replace('\t', '_T') \
         .replace(' ', '_S')
 
     return test_name or 'test-with-empty-name'

rhodecode/tests/plugin.py

@@ -1,1832 +1,1832 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 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 collections
 import datetime
 import hashlib
 import os
 import re
 import pprint
 import shutil
 import socket
 import subprocess32
 import time
 import uuid
 import dateutil.tz
 
 import mock
 import pyramid.testing
 import pytest
 import colander
 import requests
 
 import rhodecode
 from rhodecode.lib.utils2 import AttributeDict
 from rhodecode.model.changeset_status import ChangesetStatusModel
 from rhodecode.model.comment import CommentsModel
 from rhodecode.model.db import (
     PullRequest, Repository, RhodeCodeSetting, ChangesetStatus, RepoGroup,
     UserGroup, RepoRhodeCodeUi, RepoRhodeCodeSetting, RhodeCodeUi)
 from rhodecode.model.meta import Session
 from rhodecode.model.pull_request import PullRequestModel
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.user import UserModel
 from rhodecode.model.settings import VcsSettingsModel
 from rhodecode.model.user_group import UserGroupModel
 from rhodecode.model.integration import IntegrationModel
 from rhodecode.integrations import integration_type_registry
 from rhodecode.integrations.types.base import IntegrationTypeBase
 from rhodecode.lib.utils import repo2db_mapper
 from rhodecode.lib.vcs import create_vcsserver_proxy
 from rhodecode.lib.vcs.backends import get_backend
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.tests import (
     login_user_session, get_new_dir, utils, TESTS_TMP_PATH,
     TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR2_LOGIN,
     TEST_USER_REGULAR_PASS)
 from rhodecode.tests.utils import CustomTestApp, set_anonymous_access, add_test_routes
 from rhodecode.tests.fixture import Fixture
 
 
 def _split_comma(value):
     return value.split(',')
 
 
 def pytest_addoption(parser):
     parser.addoption(
         '--keep-tmp-path', action='store_true',
         help="Keep the test temporary directories")
     parser.addoption(
         '--backends', action='store', type=_split_comma,
         default=['git', 'hg', 'svn'],
         help="Select which backends to test for backend specific tests.")
     parser.addoption(
         '--dbs', action='store', type=_split_comma,
         default=['sqlite'],
         help="Select which database to test for database specific tests. "
              "Possible options are sqlite,postgres,mysql")
     parser.addoption(
         '--appenlight', '--ae', action='store_true',
         help="Track statistics in appenlight.")
     parser.addoption(
         '--appenlight-api-key', '--ae-key',
         help="API key for Appenlight.")
     parser.addoption(
         '--appenlight-url', '--ae-url',
         default="https://ae.rhodecode.com",
         help="Appenlight service URL, defaults to https://ae.rhodecode.com")
     parser.addoption(
         '--sqlite-connection-string', action='store',
         default='', help="Connection string for the dbs tests with SQLite")
     parser.addoption(
         '--postgres-connection-string', action='store',
         default='', help="Connection string for the dbs tests with Postgres")
     parser.addoption(
         '--mysql-connection-string', action='store',
         default='', help="Connection string for the dbs tests with MySQL")
     parser.addoption(
         '--repeat', type=int, default=100,
         help="Number of repetitions in performance tests.")
 
 
 def pytest_configure(config):
     # Appy the kombu patch early on, needed for test discovery on Python 2.7.11
     from rhodecode.config import patches
     patches.kombu_1_5_1_python_2_7_11()
 
 
 def pytest_collection_modifyitems(session, config, items):
     # nottest marked, compare nose, used for transition from nose to pytest
     remaining = [
         i for i in items if getattr(i.obj, '__test__', True)]
     items[:] = remaining
 
 
 def pytest_generate_tests(metafunc):
     # Support test generation based on --backend parameter
     if 'backend_alias' in metafunc.fixturenames:
         backends = get_backends_from_metafunc(metafunc)
         scope = None
         if not backends:
             pytest.skip("Not enabled for any of selected backends")
         metafunc.parametrize('backend_alias', backends, scope=scope)
     elif hasattr(metafunc.function, 'backends'):
         backends = get_backends_from_metafunc(metafunc)
         if not backends:
             pytest.skip("Not enabled for any of selected backends")
 
 
 def get_backends_from_metafunc(metafunc):
     requested_backends = set(metafunc.config.getoption('--backends'))
     if hasattr(metafunc.function, 'backends'):
         # Supported backends by this test function, created from
         # pytest.mark.backends
         backends = metafunc.function.backends.args
     elif hasattr(metafunc.cls, 'backend_alias'):
         # Support class attribute "backend_alias", this is mainly
         # for legacy reasons for tests not yet using pytest.mark.backends
         backends = [metafunc.cls.backend_alias]
     else:
         backends = metafunc.config.getoption('--backends')
     return requested_backends.intersection(backends)
 
 
 @pytest.fixture(scope='session', autouse=True)
 def activate_example_rcextensions(request):
     """
     Patch in an example rcextensions module which verifies passed in kwargs.
     """
     from rhodecode.tests.other import example_rcextensions
 
     old_extensions = rhodecode.EXTENSIONS
     rhodecode.EXTENSIONS = example_rcextensions
 
     @request.addfinalizer
     def cleanup():
         rhodecode.EXTENSIONS = old_extensions
 
 
 @pytest.fixture
 def capture_rcextensions():
     """
     Returns the recorded calls to entry points in rcextensions.
     """
     calls = rhodecode.EXTENSIONS.calls
     calls.clear()
     # Note: At this moment, it is still the empty dict, but that will
     # be filled during the test run and since it is a reference this
     # is enough to make it work.
     return calls
 
 
 @pytest.fixture(scope='session')
 def http_environ_session():
     """
     Allow to use "http_environ" in session scope.
     """
     return http_environ(
         http_host_stub=http_host_stub())
 
 
 @pytest.fixture
 def http_host_stub():
     """
     Value of HTTP_HOST in the test run.
     """
     return 'example.com:80'
 
 
 @pytest.fixture
 def http_host_only_stub():
     """
     Value of HTTP_HOST in the test run.
     """
     return http_host_stub().split(':')[0]
 
 
 @pytest.fixture
 def http_environ(http_host_stub):
     """
     HTTP extra environ keys.
 
     User by the test application and as well for setting up the pylons
     environment. In the case of the fixture "app" it should be possible
     to override this for a specific test case.
     """
     return {
         'SERVER_NAME': http_host_only_stub(),
         'SERVER_PORT': http_host_stub.split(':')[1],
         'HTTP_HOST': http_host_stub,
         'HTTP_USER_AGENT': 'rc-test-agent',
         'REQUEST_METHOD': 'GET'
     }
 
 
 @pytest.fixture(scope='function')
 def app(request, config_stub, pylonsapp, http_environ):
     app = CustomTestApp(
         pylonsapp,
         extra_environ=http_environ)
     if request.cls:
         request.cls.app = app
     return app
 
 
 @pytest.fixture(scope='session')
 def app_settings(pylonsapp, pylons_config):
     """
     Settings dictionary used to create the app.
 
     Parses the ini file and passes the result through the sanitize and apply
     defaults mechanism in `rhodecode.config.middleware`.
     """
     from paste.deploy.loadwsgi import loadcontext, APP
     from rhodecode.config.middleware import (
         sanitize_settings_and_apply_defaults)
     context = loadcontext(APP, 'config:' + pylons_config)
     settings = sanitize_settings_and_apply_defaults(context.config())
     return settings
 
 
 @pytest.fixture(scope='session')
 def db(app_settings):
     """
     Initializes the database connection.
 
     It uses the same settings which are used to create the ``pylonsapp`` or
     ``app`` fixtures.
     """
     from rhodecode.config.utils import initialize_database
     initialize_database(app_settings)
 
 
 LoginData = collections.namedtuple('LoginData', ('csrf_token', 'user'))
 
 
 def _autologin_user(app, *args):
     session = login_user_session(app, *args)
     csrf_token = rhodecode.lib.auth.get_csrf_token(session)
     return LoginData(csrf_token, session['rhodecode_user'])
 
 
 @pytest.fixture
 def autologin_user(app):
     """
     Utility fixture which makes sure that the admin user is logged in
     """
     return _autologin_user(app)
 
 
 @pytest.fixture
 def autologin_regular_user(app):
     """
     Utility fixture which makes sure that the regular user is logged in
     """
     return _autologin_user(
         app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
 
 
 @pytest.fixture(scope='function')
 def csrf_token(request, autologin_user):
     return autologin_user.csrf_token
 
 
 @pytest.fixture(scope='function')
 def xhr_header(request):
     return {'HTTP_X_REQUESTED_WITH': 'XMLHttpRequest'}
 
 
 @pytest.fixture
 def real_crypto_backend(monkeypatch):
     """
     Switch the production crypto backend on for this test.
 
     During the test run the crypto backend is replaced with a faster
     implementation based on the MD5 algorithm.
     """
     monkeypatch.setattr(rhodecode, 'is_test', False)
 
 
 @pytest.fixture(scope='class')
 def index_location(request, pylonsapp):
     index_location = pylonsapp.config['app_conf']['search.location']
     if request.cls:
         request.cls.index_location = index_location
     return index_location
 
 
 @pytest.fixture(scope='session', autouse=True)
 def tests_tmp_path(request):
     """
     Create temporary directory to be used during the test session.
     """
     if not os.path.exists(TESTS_TMP_PATH):
         os.makedirs(TESTS_TMP_PATH)
 
     if not request.config.getoption('--keep-tmp-path'):
         @request.addfinalizer
         def remove_tmp_path():
             shutil.rmtree(TESTS_TMP_PATH)
 
     return TESTS_TMP_PATH
 
 
 @pytest.fixture
 def test_repo_group(request):
     """
     Create a temporary repository group, and destroy it after
     usage automatically
     """
     fixture = Fixture()
     repogroupid = 'test_repo_group_%s' % str(time.time()).replace('.', '')
     repo_group = fixture.create_repo_group(repogroupid)
 
     def _cleanup():
         fixture.destroy_repo_group(repogroupid)
 
     request.addfinalizer(_cleanup)
     return repo_group
 
 
 @pytest.fixture
 def test_user_group(request):
     """
     Create a temporary user group, and destroy it after
     usage automatically
     """
     fixture = Fixture()
     usergroupid = 'test_user_group_%s' % str(time.time()).replace('.', '')
     user_group = fixture.create_user_group(usergroupid)
 
     def _cleanup():
         fixture.destroy_user_group(user_group)
 
     request.addfinalizer(_cleanup)
     return user_group
 
 
 @pytest.fixture(scope='session')
 def test_repo(request):
     container = TestRepoContainer()
     request.addfinalizer(container._cleanup)
     return container
 
 
 class TestRepoContainer(object):
     """
     Container for test repositories which are used read only.
 
     Repositories will be created on demand and re-used during the lifetime
     of this object.
 
     Usage to get the svn test repository "minimal"::
 
        test_repo = TestContainer()
        repo = test_repo('minimal', 'svn')
 
     """
 
     dump_extractors = {
         'git': utils.extract_git_repo_from_dump,
         'hg': utils.extract_hg_repo_from_dump,
         'svn': utils.extract_svn_repo_from_dump,
     }
 
     def __init__(self):
         self._cleanup_repos = []
         self._fixture = Fixture()
         self._repos = {}
 
     def __call__(self, dump_name, backend_alias, config=None):
         key = (dump_name, backend_alias)
         if key not in self._repos:
             repo = self._create_repo(dump_name, backend_alias, config)
             self._repos[key] = repo.repo_id
         return Repository.get(self._repos[key])
 
     def _create_repo(self, dump_name, backend_alias, config):
         repo_name = '%s-%s' % (backend_alias, dump_name)
         backend_class = get_backend(backend_alias)
         dump_extractor = self.dump_extractors[backend_alias]
         repo_path = dump_extractor(dump_name, repo_name)
 
         vcs_repo = backend_class(repo_path, config=config)
         repo2db_mapper({repo_name: vcs_repo})
 
         repo = RepoModel().get_by_repo_name(repo_name)
         self._cleanup_repos.append(repo_name)
         return repo
 
     def _cleanup(self):
         for repo_name in reversed(self._cleanup_repos):
             self._fixture.destroy_repo(repo_name)
 
 
 @pytest.fixture
 def backend(request, backend_alias, pylonsapp, test_repo):
     """
     Parametrized fixture which represents a single backend implementation.
 
     It respects the option `--backends` to focus the test run on specific
     backend implementations.
 
     It also supports `pytest.mark.xfail_backends` to mark tests as failing
     for specific backends. This is intended as a utility for incremental
     development of a new backend implementation.
     """
     if backend_alias not in request.config.getoption('--backends'):
         pytest.skip("Backend %s not selected." % (backend_alias, ))
 
     utils.check_xfail_backends(request.node, backend_alias)
     utils.check_skip_backends(request.node, backend_alias)
 
     repo_name = 'vcs_test_%s' % (backend_alias, )
     backend = Backend(
         alias=backend_alias,
         repo_name=repo_name,
         test_name=request.node.name,
         test_repo_container=test_repo)
     request.addfinalizer(backend.cleanup)
     return backend
 
 
 @pytest.fixture
 def backend_git(request, pylonsapp, test_repo):
     return backend(request, 'git', pylonsapp, test_repo)
 
 
 @pytest.fixture
 def backend_hg(request, pylonsapp, test_repo):
     return backend(request, 'hg', pylonsapp, test_repo)
 
 
 @pytest.fixture
 def backend_svn(request, pylonsapp, test_repo):
     return backend(request, 'svn', pylonsapp, test_repo)
 
 
 @pytest.fixture
 def backend_random(backend_git):
     """
     Use this to express that your tests need "a backend.
 
     A few of our tests need a backend, so that we can run the code. This
     fixture is intended to be used for such cases. It will pick one of the
     backends and run the tests.
 
     The fixture `backend` would run the test multiple times for each
     available backend which is a pure waste of time if the test is
     independent of the backend type.
     """
     # TODO: johbo: Change this to pick a random backend
     return backend_git
 
 
 @pytest.fixture
 def backend_stub(backend_git):
     """
     Use this to express that your tests need a backend stub
 
     TODO: mikhail: Implement a real stub logic instead of returning
     a git backend
     """
     return backend_git
 
 
 @pytest.fixture
 def repo_stub(backend_stub):
     """
     Use this to express that your tests need a repository stub
     """
     return backend_stub.create_repo()
 
 
 class Backend(object):
     """
     Represents the test configuration for one supported backend
 
     Provides easy access to different test repositories based on
     `__getitem__`. Such repositories will only be created once per test
     session.
     """
 
     invalid_repo_name = re.compile(r'[^0-9a-zA-Z]+')
     _master_repo = None
     _commit_ids = {}
 
     def __init__(self, alias, repo_name, test_name, test_repo_container):
         self.alias = alias
         self.repo_name = repo_name
         self._cleanup_repos = []
         self._test_name = test_name
         self._test_repo_container = test_repo_container
         # TODO: johbo: Used as a delegate interim. Not yet sure if Backend or
         # Fixture will survive in the end.
         self._fixture = Fixture()
 
     def __getitem__(self, key):
         return self._test_repo_container(key, self.alias)
 
     def create_test_repo(self, key, config=None):
         return self._test_repo_container(key, self.alias, config)
 
     @property
     def repo(self):
         """
         Returns the "current" repository. This is the vcs_test repo or the
         last repo which has been created with `create_repo`.
         """
         from rhodecode.model.db import Repository
         return Repository.get_by_repo_name(self.repo_name)
 
     @property
     def default_branch_name(self):
         VcsRepository = get_backend(self.alias)
         return VcsRepository.DEFAULT_BRANCH_NAME
 
     @property
     def default_head_id(self):
         """
         Returns the default head id of the underlying backend.
 
         This will be the default branch name in case the backend does have a
         default branch. In the other cases it will point to a valid head
         which can serve as the base to create a new commit on top of it.
         """
         vcsrepo = self.repo.scm_instance()
         head_id = (
             vcsrepo.DEFAULT_BRANCH_NAME or
             vcsrepo.commit_ids[-1])
         return head_id
 
     @property
     def commit_ids(self):
         """
         Returns the list of commits for the last created repository
         """
         return self._commit_ids
 
     def create_master_repo(self, commits):
         """
         Create a repository and remember it as a template.
 
         This allows to easily create derived repositories to construct
         more complex scenarios for diff, compare and pull requests.
 
         Returns a commit map which maps from commit message to raw_id.
         """
         self._master_repo = self.create_repo(commits=commits)
         return self._commit_ids
 
     def create_repo(
             self, commits=None, number_of_commits=0, heads=None,
             name_suffix=u'', **kwargs):
         """
         Create a repository and record it for later cleanup.
 
         :param commits: Optional. A sequence of dict instances.
            Will add a commit per entry to the new repository.
         :param number_of_commits: Optional. If set to a number, this number of
            commits will be added to the new repository.
         :param heads: Optional. Can be set to a sequence of of commit
            names which shall be pulled in from the master repository.
 
         """
         self.repo_name = self._next_repo_name() + name_suffix
         repo = self._fixture.create_repo(
             self.repo_name, repo_type=self.alias, **kwargs)
         self._cleanup_repos.append(repo.repo_name)
 
         commits = commits or [
             {'message': 'Commit %s of %s' % (x, self.repo_name)}
             for x in xrange(number_of_commits)]
         self._add_commits_to_repo(repo.scm_instance(), commits)
         if heads:
             self.pull_heads(repo, heads)
 
         return repo
 
     def pull_heads(self, repo, heads):
         """
         Make sure that repo contains all commits mentioned in `heads`
         """
         vcsmaster = self._master_repo.scm_instance()
         vcsrepo = repo.scm_instance()
         vcsrepo.config.clear_section('hooks')
         commit_ids = [self._commit_ids[h] for h in heads]
         vcsrepo.pull(vcsmaster.path, commit_ids=commit_ids)
 
     def create_fork(self):
         repo_to_fork = self.repo_name
         self.repo_name = self._next_repo_name()
         repo = self._fixture.create_fork(repo_to_fork, self.repo_name)
         self._cleanup_repos.append(self.repo_name)
         return repo
 
     def new_repo_name(self, suffix=u''):
         self.repo_name = self._next_repo_name() + suffix
         self._cleanup_repos.append(self.repo_name)
         return self.repo_name
 
     def _next_repo_name(self):
         return u"%s_%s" % (
             self.invalid_repo_name.sub(u'_', self._test_name),
             len(self._cleanup_repos))
 
     def ensure_file(self, filename, content='Test content\n'):
         assert self._cleanup_repos, "Avoid writing into vcs_test repos"
         commits = [
             {'added': [
                 FileNode(filename, content=content),
             ]},
         ]
         self._add_commits_to_repo(self.repo.scm_instance(), commits)
 
     def enable_downloads(self):
         repo = self.repo
         repo.enable_downloads = True
         Session().add(repo)
         Session().commit()
 
     def cleanup(self):
         for repo_name in reversed(self._cleanup_repos):
             self._fixture.destroy_repo(repo_name)
 
     def _add_commits_to_repo(self, repo, commits):
         commit_ids = _add_commits_to_repo(repo, commits)
         if not commit_ids:
             return
         self._commit_ids = commit_ids
 
         # Creating refs for Git to allow fetching them from remote repository
         if self.alias == 'git':
             refs = {}
             for message in self._commit_ids:
                 # TODO: mikhail: do more special chars replacements
                 ref_name = 'refs/test-refs/{}'.format(
                     message.replace(' ', ''))
                 refs[ref_name] = self._commit_ids[message]
             self._create_refs(repo, refs)
 
     def _create_refs(self, repo, refs):
         for ref_name in refs:
             repo.set_refs(ref_name, refs[ref_name])
 
 
 @pytest.fixture
 def vcsbackend(request, backend_alias, tests_tmp_path, pylonsapp, test_repo):
     """
     Parametrized fixture which represents a single vcs backend implementation.
 
     See the fixture `backend` for more details. This one implements the same
     concept, but on vcs level. So it does not provide model instances etc.
 
     Parameters are generated dynamically, see :func:`pytest_generate_tests`
     for how this works.
     """
     if backend_alias not in request.config.getoption('--backends'):
         pytest.skip("Backend %s not selected." % (backend_alias, ))
 
     utils.check_xfail_backends(request.node, backend_alias)
     utils.check_skip_backends(request.node, backend_alias)
 
     repo_name = 'vcs_test_%s' % (backend_alias, )
     repo_path = os.path.join(tests_tmp_path, repo_name)
     backend = VcsBackend(
         alias=backend_alias,
         repo_path=repo_path,
         test_name=request.node.name,
         test_repo_container=test_repo)
     request.addfinalizer(backend.cleanup)
     return backend
 
 
 @pytest.fixture
 def vcsbackend_git(request, tests_tmp_path, pylonsapp, test_repo):
     return vcsbackend(request, 'git', tests_tmp_path, pylonsapp, test_repo)
 
 
 @pytest.fixture
 def vcsbackend_hg(request, tests_tmp_path, pylonsapp, test_repo):
     return vcsbackend(request, 'hg', tests_tmp_path, pylonsapp, test_repo)
 
 
 @pytest.fixture
 def vcsbackend_svn(request, tests_tmp_path, pylonsapp, test_repo):
     return vcsbackend(request, 'svn', tests_tmp_path, pylonsapp, test_repo)
 
 
 @pytest.fixture
 def vcsbackend_random(vcsbackend_git):
     """
     Use this to express that your tests need "a vcsbackend".
 
     The fixture `vcsbackend` would run the test multiple times for each
     available vcs backend which is a pure waste of time if the test is
     independent of the vcs backend type.
     """
     # TODO: johbo: Change this to pick a random backend
     return vcsbackend_git
 
 
 @pytest.fixture
 def vcsbackend_stub(vcsbackend_git):
     """
     Use this to express that your test just needs a stub of a vcsbackend.
 
     Plan is to eventually implement an in-memory stub to speed tests up.
     """
     return vcsbackend_git
 
 
 class VcsBackend(object):
     """
     Represents the test configuration for one supported vcs backend.
     """
 
     invalid_repo_name = re.compile(r'[^0-9a-zA-Z]+')
 
     def __init__(self, alias, repo_path, test_name, test_repo_container):
         self.alias = alias
         self._repo_path = repo_path
         self._cleanup_repos = []
         self._test_name = test_name
         self._test_repo_container = test_repo_container
 
     def __getitem__(self, key):
         return self._test_repo_container(key, self.alias).scm_instance()
 
     @property
     def repo(self):
         """
         Returns the "current" repository. This is the vcs_test repo of the last
         repo which has been created.
         """
         Repository = get_backend(self.alias)
         return Repository(self._repo_path)
 
     @property
     def backend(self):
         """
         Returns the backend implementation class.
         """
         return get_backend(self.alias)
 
     def create_repo(self, commits=None, number_of_commits=0, _clone_repo=None):
         repo_name = self._next_repo_name()
         self._repo_path = get_new_dir(repo_name)
         repo_class = get_backend(self.alias)
         src_url = None
         if _clone_repo:
             src_url = _clone_repo.path
         repo = repo_class(self._repo_path, create=True, src_url=src_url)
         self._cleanup_repos.append(repo)
 
         commits = commits or [
             {'message': 'Commit %s of %s' % (x, repo_name)}
             for x in xrange(number_of_commits)]
         _add_commits_to_repo(repo, commits)
         return repo
 
     def clone_repo(self, repo):
         return self.create_repo(_clone_repo=repo)
 
     def cleanup(self):
         for repo in self._cleanup_repos:
             shutil.rmtree(repo.path)
 
     def new_repo_path(self):
         repo_name = self._next_repo_name()
         self._repo_path = get_new_dir(repo_name)
         return self._repo_path
 
     def _next_repo_name(self):
         return "%s_%s" % (
             self.invalid_repo_name.sub('_', self._test_name),
             len(self._cleanup_repos))
 
     def add_file(self, repo, filename, content='Test content\n'):
         imc = repo.in_memory_commit
         imc.add(FileNode(filename, content=content))
         imc.commit(
             message=u'Automatic commit from vcsbackend fixture',
             author=u'Automatic')
 
     def ensure_file(self, filename, content='Test content\n'):
         assert self._cleanup_repos, "Avoid writing into vcs_test repos"
         self.add_file(self.repo, filename, content)
 
 
 def _add_commits_to_repo(vcs_repo, commits):
     commit_ids = {}
     if not commits:
         return commit_ids
 
     imc = vcs_repo.in_memory_commit
     commit = None
 
     for idx, commit in enumerate(commits):
         message = unicode(commit.get('message', 'Commit %s' % idx))
 
         for node in commit.get('added', []):
             imc.add(FileNode(node.path, content=node.content))
         for node in commit.get('changed', []):
             imc.change(FileNode(node.path, content=node.content))
         for node in commit.get('removed', []):
             imc.remove(FileNode(node.path))
 
         parents = [
             vcs_repo.get_commit(commit_id=commit_ids[p])
             for p in commit.get('parents', [])]
 
         operations = ('added', 'changed', 'removed')
         if not any((commit.get(o) for o in operations)):
             imc.add(FileNode('file_%s' % idx, content=message))
 
         commit = imc.commit(
             message=message,
             author=unicode(commit.get('author', 'Automatic')),
             date=commit.get('date'),
             branch=commit.get('branch'),
             parents=parents)
 
         commit_ids[commit.message] = commit.raw_id
 
     return commit_ids
 
 
 @pytest.fixture
 def reposerver(request):
     """
     Allows to serve a backend repository
     """
 
     repo_server = RepoServer()
     request.addfinalizer(repo_server.cleanup)
     return repo_server
 
 
 class RepoServer(object):
     """
     Utility to serve a local repository for the duration of a test case.
 
     Supports only Subversion so far.
     """
 
     url = None
 
     def __init__(self):
         self._cleanup_servers = []
 
     def serve(self, vcsrepo):
         if vcsrepo.alias != 'svn':
             raise TypeError("Backend %s not supported" % vcsrepo.alias)
 
         proc = subprocess32.Popen(
             ['svnserve', '-d', '--foreground', '--listen-host', 'localhost',
              '--root', vcsrepo.path])
         self._cleanup_servers.append(proc)
         self.url = 'svn://localhost'
 
     def cleanup(self):
         for proc in self._cleanup_servers:
             proc.terminate()
 
 
 @pytest.fixture
 def pr_util(backend, request, config_stub):
     """
     Utility for tests of models and for functional tests around pull requests.
 
     It gives an instance of :class:`PRTestUtility` which provides various
     utility methods around one pull request.
 
     This fixture uses `backend` and inherits its parameterization.
     """
 
     util = PRTestUtility(backend)
 
     @request.addfinalizer
     def cleanup():
         util.cleanup()
 
     return util
 
 
 class PRTestUtility(object):
 
     pull_request = None
     pull_request_id = None
     mergeable_patcher = None
     mergeable_mock = None
     notification_patcher = None
 
     def __init__(self, backend):
         self.backend = backend
 
     def create_pull_request(
             self, commits=None, target_head=None, source_head=None,
             revisions=None, approved=False, author=None, mergeable=False,
             enable_notifications=True, name_suffix=u'', reviewers=None,
             title=u"Test", description=u"Description"):
         self.set_mergeable(mergeable)
         if not enable_notifications:
             # mock notification side effect
             self.notification_patcher = mock.patch(
                 'rhodecode.model.notification.NotificationModel.create')
             self.notification_patcher.start()
 
         if not self.pull_request:
             if not commits:
                 commits = [
                     {'message': 'c1'},
                     {'message': 'c2'},
                     {'message': 'c3'},
                 ]
                 target_head = 'c1'
                 source_head = 'c2'
                 revisions = ['c2']
 
             self.commit_ids = self.backend.create_master_repo(commits)
             self.target_repository = self.backend.create_repo(
                 heads=[target_head], name_suffix=name_suffix)
             self.source_repository = self.backend.create_repo(
                 heads=[source_head], name_suffix=name_suffix)
             self.author = author or UserModel().get_by_username(
                 TEST_USER_ADMIN_LOGIN)
 
             model = PullRequestModel()
             self.create_parameters = {
                 'created_by': self.author,
                 'source_repo': self.source_repository.repo_name,
                 'source_ref': self._default_branch_reference(source_head),
                 'target_repo': self.target_repository.repo_name,
                 'target_ref': self._default_branch_reference(target_head),
                 'revisions': [self.commit_ids[r] for r in revisions],
                 'reviewers': reviewers or self._get_reviewers(),
                 'title': title,
                 'description': description,
             }
             self.pull_request = model.create(**self.create_parameters)
             assert model.get_versions(self.pull_request) == []
 
             self.pull_request_id = self.pull_request.pull_request_id
 
             if approved:
                 self.approve()
 
             Session().add(self.pull_request)
             Session().commit()
 
         return self.pull_request
 
     def approve(self):
         self.create_status_votes(
             ChangesetStatus.STATUS_APPROVED,
             *self.pull_request.reviewers)
 
     def close(self):
         PullRequestModel().close_pull_request(self.pull_request, self.author)
 
     def _default_branch_reference(self, commit_message):
         reference = '%s:%s:%s' % (
             'branch',
             self.backend.default_branch_name,
             self.commit_ids[commit_message])
         return reference
 
     def _get_reviewers(self):
         return [
             (TEST_USER_REGULAR_LOGIN, ['default1'], False),
             (TEST_USER_REGULAR2_LOGIN, ['default2'], False),
         ]
 
     def update_source_repository(self, head=None):
         heads = [head or 'c3']
         self.backend.pull_heads(self.source_repository, heads=heads)
 
     def add_one_commit(self, head=None):
         self.update_source_repository(head=head)
         old_commit_ids = set(self.pull_request.revisions)
         PullRequestModel().update_commits(self.pull_request)
         commit_ids = set(self.pull_request.revisions)
         new_commit_ids = commit_ids - old_commit_ids
         assert len(new_commit_ids) == 1
         return new_commit_ids.pop()
 
     def remove_one_commit(self):
         assert len(self.pull_request.revisions) == 2
         source_vcs = self.source_repository.scm_instance()
         removed_commit_id = source_vcs.commit_ids[-1]
 
         # TODO: johbo: Git and Mercurial have an inconsistent vcs api here,
         # remove the if once that's sorted out.
         if self.backend.alias == "git":
             kwargs = {'branch_name': self.backend.default_branch_name}
         else:
             kwargs = {}
         source_vcs.strip(removed_commit_id, **kwargs)
 
         PullRequestModel().update_commits(self.pull_request)
         assert len(self.pull_request.revisions) == 1
         return removed_commit_id
 
     def create_comment(self, linked_to=None):
         comment = CommentsModel().create(
             text=u"Test comment",
             repo=self.target_repository.repo_name,
             user=self.author,
             pull_request=self.pull_request)
         assert comment.pull_request_version_id is None
 
         if linked_to:
             PullRequestModel()._link_comments_to_version(linked_to)
 
         return comment
 
     def create_inline_comment(
             self, linked_to=None, line_no=u'n1', file_path='file_1'):
         comment = CommentsModel().create(
             text=u"Test comment",
             repo=self.target_repository.repo_name,
             user=self.author,
             line_no=line_no,
             f_path=file_path,
             pull_request=self.pull_request)
         assert comment.pull_request_version_id is None
 
         if linked_to:
             PullRequestModel()._link_comments_to_version(linked_to)
 
         return comment
 
     def create_version_of_pull_request(self):
         pull_request = self.create_pull_request()
         version = PullRequestModel()._create_version_from_snapshot(
             pull_request)
         return version
 
     def create_status_votes(self, status, *reviewers):
         for reviewer in reviewers:
             ChangesetStatusModel().set_status(
                 repo=self.pull_request.target_repo,
                 status=status,
                 user=reviewer.user_id,
                 pull_request=self.pull_request)
 
     def set_mergeable(self, value):
         if not self.mergeable_patcher:
             self.mergeable_patcher = mock.patch.object(
                 VcsSettingsModel, 'get_general_settings')
             self.mergeable_mock = self.mergeable_patcher.start()
         self.mergeable_mock.return_value = {
             'rhodecode_pr_merge_enabled': value}
 
     def cleanup(self):
         # In case the source repository is already cleaned up, the pull
         # request will already be deleted.
         pull_request = PullRequest().get(self.pull_request_id)
         if pull_request:
             PullRequestModel().delete(pull_request, pull_request.author)
             Session().commit()
 
         if self.notification_patcher:
             self.notification_patcher.stop()
 
         if self.mergeable_patcher:
             self.mergeable_patcher.stop()
 
 
 @pytest.fixture
 def user_admin(pylonsapp):
     """
     Provides the default admin test user as an instance of `db.User`.
     """
     user = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
     return user
 
 
 @pytest.fixture
 def user_regular(pylonsapp):
     """
     Provides the default regular test user as an instance of `db.User`.
     """
     user = UserModel().get_by_username(TEST_USER_REGULAR_LOGIN)
     return user
 
 
 @pytest.fixture
 def user_util(request, pylonsapp):
     """
     Provides a wired instance of `UserUtility` with integrated cleanup.
     """
     utility = UserUtility(test_name=request.node.name)
     request.addfinalizer(utility.cleanup)
     return utility
 
 
 # TODO: johbo: Split this up into utilities per domain or something similar
 class UserUtility(object):
 
     def __init__(self, test_name="test"):
         self._test_name = self._sanitize_name(test_name)
         self.fixture = Fixture()
         self.repo_group_ids = []
         self.repos_ids = []
         self.user_ids = []
         self.user_group_ids = []
         self.user_repo_permission_ids = []
         self.user_group_repo_permission_ids = []
         self.user_repo_group_permission_ids = []
         self.user_group_repo_group_permission_ids = []
         self.user_user_group_permission_ids = []
         self.user_group_user_group_permission_ids = []
         self.user_permissions = []
 
     def _sanitize_name(self, name):
         for char in ['[', ']']:
             name = name.replace(char, '_')
         return name
 
     def create_repo_group(
             self, owner=TEST_USER_ADMIN_LOGIN, auto_cleanup=True):
         group_name = "{prefix}_repogroup_{count}".format(
             prefix=self._test_name,
             count=len(self.repo_group_ids))
         repo_group = self.fixture.create_repo_group(
             group_name, cur_user=owner)
         if auto_cleanup:
             self.repo_group_ids.append(repo_group.group_id)
         return repo_group
 
     def create_repo(self, owner=TEST_USER_ADMIN_LOGIN, parent=None,
                     auto_cleanup=True, repo_type='hg'):
         repo_name = "{prefix}_repository_{count}".format(
             prefix=self._test_name,
             count=len(self.repos_ids))
 
         repository = self.fixture.create_repo(
             repo_name, cur_user=owner, repo_group=parent, repo_type=repo_type)
         if auto_cleanup:
             self.repos_ids.append(repository.repo_id)
         return repository
 
     def create_user(self, auto_cleanup=True, **kwargs):
         user_name = "{prefix}_user_{count}".format(
             prefix=self._test_name,
             count=len(self.user_ids))
         user = self.fixture.create_user(user_name, **kwargs)
         if auto_cleanup:
             self.user_ids.append(user.user_id)
         return user
 
     def create_user_with_group(self):
         user = self.create_user()
         user_group = self.create_user_group(members=[user])
         return user, user_group
 
     def create_user_group(self, owner=TEST_USER_ADMIN_LOGIN, members=None,
                           auto_cleanup=True, **kwargs):
         group_name = "{prefix}_usergroup_{count}".format(
             prefix=self._test_name,
             count=len(self.user_group_ids))
         user_group = self.fixture.create_user_group(
             group_name, cur_user=owner, **kwargs)
 
         if auto_cleanup:
             self.user_group_ids.append(user_group.users_group_id)
         if members:
             for user in members:
                 UserGroupModel().add_user_to_group(user_group, user)
         return user_group
 
     def grant_user_permission(self, user_name, permission_name):
         self._inherit_default_user_permissions(user_name, False)
         self.user_permissions.append((user_name, permission_name))
 
     def grant_user_permission_to_repo_group(
             self, repo_group, user, permission_name):
         permission = RepoGroupModel().grant_user_permission(
             repo_group, user, permission_name)
         self.user_repo_group_permission_ids.append(
             (repo_group.group_id, user.user_id))
         return permission
 
     def grant_user_group_permission_to_repo_group(
             self, repo_group, user_group, permission_name):
         permission = RepoGroupModel().grant_user_group_permission(
             repo_group, user_group, permission_name)
         self.user_group_repo_group_permission_ids.append(
             (repo_group.group_id, user_group.users_group_id))
         return permission
 
     def grant_user_permission_to_repo(
             self, repo, user, permission_name):
         permission = RepoModel().grant_user_permission(
             repo, user, permission_name)
         self.user_repo_permission_ids.append(
             (repo.repo_id, user.user_id))
         return permission
 
     def grant_user_group_permission_to_repo(
             self, repo, user_group, permission_name):
         permission = RepoModel().grant_user_group_permission(
             repo, user_group, permission_name)
         self.user_group_repo_permission_ids.append(
             (repo.repo_id, user_group.users_group_id))
         return permission
 
     def grant_user_permission_to_user_group(
             self, target_user_group, user, permission_name):
         permission = UserGroupModel().grant_user_permission(
             target_user_group, user, permission_name)
         self.user_user_group_permission_ids.append(
             (target_user_group.users_group_id, user.user_id))
         return permission
 
     def grant_user_group_permission_to_user_group(
             self, target_user_group, user_group, permission_name):
         permission = UserGroupModel().grant_user_group_permission(
             target_user_group, user_group, permission_name)
         self.user_group_user_group_permission_ids.append(
             (target_user_group.users_group_id, user_group.users_group_id))
         return permission
 
     def revoke_user_permission(self, user_name, permission_name):
         self._inherit_default_user_permissions(user_name, True)
         UserModel().revoke_perm(user_name, permission_name)
 
     def _inherit_default_user_permissions(self, user_name, value):
         user = UserModel().get_by_username(user_name)
         user.inherit_default_permissions = value
         Session().add(user)
         Session().commit()
 
     def cleanup(self):
         self._cleanup_permissions()
         self._cleanup_repos()
         self._cleanup_repo_groups()
         self._cleanup_user_groups()
         self._cleanup_users()
 
     def _cleanup_permissions(self):
         if self.user_permissions:
             for user_name, permission_name in self.user_permissions:
                 self.revoke_user_permission(user_name, permission_name)
 
         for permission in self.user_repo_permission_ids:
             RepoModel().revoke_user_permission(*permission)
 
         for permission in self.user_group_repo_permission_ids:
             RepoModel().revoke_user_group_permission(*permission)
 
         for permission in self.user_repo_group_permission_ids:
             RepoGroupModel().revoke_user_permission(*permission)
 
         for permission in self.user_group_repo_group_permission_ids:
             RepoGroupModel().revoke_user_group_permission(*permission)
 
         for permission in self.user_user_group_permission_ids:
             UserGroupModel().revoke_user_permission(*permission)
 
         for permission in self.user_group_user_group_permission_ids:
             UserGroupModel().revoke_user_group_permission(*permission)
 
     def _cleanup_repo_groups(self):
         def _repo_group_compare(first_group_id, second_group_id):
             """
             Gives higher priority to the groups with the most complex paths
             """
             first_group = RepoGroup.get(first_group_id)
             second_group = RepoGroup.get(second_group_id)
             first_group_parts = (
                 len(first_group.group_name.split('/')) if first_group else 0)
             second_group_parts = (
                 len(second_group.group_name.split('/')) if second_group else 0)
             return cmp(second_group_parts, first_group_parts)
 
         sorted_repo_group_ids = sorted(
             self.repo_group_ids, cmp=_repo_group_compare)
         for repo_group_id in sorted_repo_group_ids:
             self.fixture.destroy_repo_group(repo_group_id)
 
     def _cleanup_repos(self):
         sorted_repos_ids = sorted(self.repos_ids)
         for repo_id in sorted_repos_ids:
             self.fixture.destroy_repo(repo_id)
 
     def _cleanup_user_groups(self):
         def _user_group_compare(first_group_id, second_group_id):
             """
             Gives higher priority to the groups with the most complex paths
             """
             first_group = UserGroup.get(first_group_id)
             second_group = UserGroup.get(second_group_id)
             first_group_parts = (
                 len(first_group.users_group_name.split('/'))
                 if first_group else 0)
             second_group_parts = (
                 len(second_group.users_group_name.split('/'))
                 if second_group else 0)
             return cmp(second_group_parts, first_group_parts)
 
         sorted_user_group_ids = sorted(
             self.user_group_ids, cmp=_user_group_compare)
         for user_group_id in sorted_user_group_ids:
             self.fixture.destroy_user_group(user_group_id)
 
     def _cleanup_users(self):
         for user_id in self.user_ids:
             self.fixture.destroy_user(user_id)
 
 
 # TODO: Think about moving this into a pytest-pyro package and make it a
 # pytest plugin
 @pytest.hookimpl(tryfirst=True, hookwrapper=True)
 def pytest_runtest_makereport(item, call):
     """
     Adding the remote traceback if the exception has this information.
 
     VCSServer attaches this information as the attribute `_vcs_server_traceback`
     to the exception instance.
     """
     outcome = yield
     report = outcome.get_result()
     if call.excinfo:
         _add_vcsserver_remote_traceback(report, call.excinfo.value)
 
 
 def _add_vcsserver_remote_traceback(report, exc):
     vcsserver_traceback = getattr(exc, '_vcs_server_traceback', None)
 
     if vcsserver_traceback:
         section = 'VCSServer remote traceback ' + report.when
         report.sections.append((section, vcsserver_traceback))
 
 
 @pytest.fixture(scope='session')
 def testrun():
     return {
         'uuid': uuid.uuid4(),
         'start': datetime.datetime.utcnow().isoformat(),
         'timestamp': int(time.time()),
     }
 
 
 @pytest.fixture(autouse=True)
 def collect_appenlight_stats(request, testrun):
     """
     This fixture reports memory consumtion of single tests.
 
     It gathers data based on `psutil` and sends them to Appenlight. The option
     ``--ae`` has te be used to enable this fixture and the API key for your
     application has to be provided in ``--ae-key``.
     """
     try:
         # cygwin cannot have yet psutil support.
         import psutil
     except ImportError:
         return
 
     if not request.config.getoption('--appenlight'):
         return
     else:
         # Only request the pylonsapp fixture if appenlight tracking is
         # enabled. This will speed up a test run of unit tests by 2 to 3
         # seconds if appenlight is not enabled.
         pylonsapp = request.getfuncargvalue("pylonsapp")
     url = '{}/api/logs'.format(request.config.getoption('--appenlight-url'))
     client = AppenlightClient(
         url=url,
         api_key=request.config.getoption('--appenlight-api-key'),
         namespace=request.node.nodeid,
         request=str(testrun['uuid']),
         testrun=testrun)
 
     client.collect({
         'message': "Starting",
     })
 
     server_and_port = pylonsapp.config['vcs.server']
     protocol = pylonsapp.config['vcs.server.protocol']
     server = create_vcsserver_proxy(server_and_port, protocol)
     with server:
         vcs_pid = server.get_pid()
         server.run_gc()
     vcs_process = psutil.Process(vcs_pid)
     mem = vcs_process.memory_info()
     client.tag_before('vcsserver.rss', mem.rss)
     client.tag_before('vcsserver.vms', mem.vms)
 
     test_process = psutil.Process()
     mem = test_process.memory_info()
     client.tag_before('test.rss', mem.rss)
     client.tag_before('test.vms', mem.vms)
 
     client.tag_before('time', time.time())
 
     @request.addfinalizer
     def send_stats():
         client.tag_after('time', time.time())
         with server:
             gc_stats = server.run_gc()
         for tag, value in gc_stats.items():
             client.tag_after(tag, value)
         mem = vcs_process.memory_info()
         client.tag_after('vcsserver.rss', mem.rss)
         client.tag_after('vcsserver.vms', mem.vms)
 
         mem = test_process.memory_info()
         client.tag_after('test.rss', mem.rss)
         client.tag_after('test.vms', mem.vms)
 
         client.collect({
             'message': "Finished",
         })
         client.send_stats()
 
     return client
 
 
 class AppenlightClient():
 
     url_template = '{url}?protocol_version=0.5'
 
     def __init__(
             self, url, api_key, add_server=True, add_timestamp=True,
             namespace=None, request=None, testrun=None):
         self.url = self.url_template.format(url=url)
         self.api_key = api_key
         self.add_server = add_server
         self.add_timestamp = add_timestamp
         self.namespace = namespace
         self.request = request
         self.server = socket.getfqdn(socket.gethostname())
         self.tags_before = {}
         self.tags_after = {}
         self.stats = []
         self.testrun = testrun or {}
 
     def tag_before(self, tag, value):
         self.tags_before[tag] = value
 
     def tag_after(self, tag, value):
         self.tags_after[tag] = value
 
     def collect(self, data):
         if self.add_server:
             data.setdefault('server', self.server)
         if self.add_timestamp:
             data.setdefault('date', datetime.datetime.utcnow().isoformat())
         if self.namespace:
             data.setdefault('namespace', self.namespace)
         if self.request:
             data.setdefault('request', self.request)
         self.stats.append(data)
 
     def send_stats(self):
         tags = [
             ('testrun', self.request),
             ('testrun.start', self.testrun['start']),
             ('testrun.timestamp', self.testrun['timestamp']),
             ('test', self.namespace),
         ]
         for key, value in self.tags_before.items():
             tags.append((key + '.before', value))
             try:
                 delta = self.tags_after[key] - value
                 tags.append((key + '.delta', delta))
             except Exception:
                 pass
         for key, value in self.tags_after.items():
             tags.append((key + '.after', value))
         self.collect({
             'message': "Collected tags",
             'tags': tags,
         })
 
         response = requests.post(
             self.url,
             headers={
                 'X-appenlight-api-key': self.api_key},
             json=self.stats,
         )
 
         if not response.status_code == 200:
             pprint.pprint(self.stats)
             print response.headers
             print response.text
             raise Exception('Sending to appenlight failed')
 
 
 @pytest.fixture
 def gist_util(request, pylonsapp):
     """
     Provides a wired instance of `GistUtility` with integrated cleanup.
     """
     utility = GistUtility()
     request.addfinalizer(utility.cleanup)
     return utility
 
 
 class GistUtility(object):
     def __init__(self):
         self.fixture = Fixture()
         self.gist_ids = []
 
     def create_gist(self, **kwargs):
         gist = self.fixture.create_gist(**kwargs)
         self.gist_ids.append(gist.gist_id)
         return gist
 
     def cleanup(self):
         for id_ in self.gist_ids:
             self.fixture.destroy_gists(str(id_))
 
 
 @pytest.fixture
 def enabled_backends(request):
     backends = request.config.option.backends
     return backends[:]
 
 
 @pytest.fixture
 def settings_util(request):
     """
     Provides a wired instance of `SettingsUtility` with integrated cleanup.
     """
     utility = SettingsUtility()
     request.addfinalizer(utility.cleanup)
     return utility
 
 
 class SettingsUtility(object):
     def __init__(self):
         self.rhodecode_ui_ids = []
         self.rhodecode_setting_ids = []
         self.repo_rhodecode_ui_ids = []
         self.repo_rhodecode_setting_ids = []
 
     def create_repo_rhodecode_ui(
             self, repo, section, value, key=None, active=True, cleanup=True):
         key = key or hashlib.sha1(
             '{}{}{}'.format(section, value, repo.repo_id)).hexdigest()
 
         setting = RepoRhodeCodeUi()
         setting.repository_id = repo.repo_id
         setting.ui_section = section
         setting.ui_value = value
         setting.ui_key = key
         setting.ui_active = active
         Session().add(setting)
         Session().commit()
 
         if cleanup:
             self.repo_rhodecode_ui_ids.append(setting.ui_id)
         return setting
 
     def create_rhodecode_ui(
             self, section, value, key=None, active=True, cleanup=True):
         key = key or hashlib.sha1('{}{}'.format(section, value)).hexdigest()
 
         setting = RhodeCodeUi()
         setting.ui_section = section
         setting.ui_value = value
         setting.ui_key = key
         setting.ui_active = active
         Session().add(setting)
         Session().commit()
 
         if cleanup:
             self.rhodecode_ui_ids.append(setting.ui_id)
         return setting
 
     def create_repo_rhodecode_setting(
             self, repo, name, value, type_, cleanup=True):
         setting = RepoRhodeCodeSetting(
             repo.repo_id, key=name, val=value, type=type_)
         Session().add(setting)
         Session().commit()
 
         if cleanup:
             self.repo_rhodecode_setting_ids.append(setting.app_settings_id)
         return setting
 
     def create_rhodecode_setting(self, name, value, type_, cleanup=True):
         setting = RhodeCodeSetting(key=name, val=value, type=type_)
         Session().add(setting)
         Session().commit()
 
         if cleanup:
             self.rhodecode_setting_ids.append(setting.app_settings_id)
 
         return setting
 
     def cleanup(self):
         for id_ in self.rhodecode_ui_ids:
             setting = RhodeCodeUi.get(id_)
             Session().delete(setting)
 
         for id_ in self.rhodecode_setting_ids:
             setting = RhodeCodeSetting.get(id_)
             Session().delete(setting)
 
         for id_ in self.repo_rhodecode_ui_ids:
             setting = RepoRhodeCodeUi.get(id_)
             Session().delete(setting)
 
         for id_ in self.repo_rhodecode_setting_ids:
             setting = RepoRhodeCodeSetting.get(id_)
             Session().delete(setting)
 
         Session().commit()
 
 
 @pytest.fixture
 def no_notifications(request):
     notification_patcher = mock.patch(
         'rhodecode.model.notification.NotificationModel.create')
     notification_patcher.start()
     request.addfinalizer(notification_patcher.stop)
 
 
 @pytest.fixture(scope='session')
 def repeat(request):
     """
     The number of repetitions is based on this fixture.
 
     Slower calls may divide it by 10 or 100. It is chosen in a way so that the
     tests are not too slow in our default test suite.
     """
     return request.config.getoption('--repeat')
 
 
 @pytest.fixture
 def rhodecode_fixtures():
     return Fixture()
 
 
 @pytest.fixture
 def request_stub():
     """
     Stub request object.
     """
-    request = pyramid.testing.DummyRequest()
-    request.scheme = 'https'
+    from rhodecode.lib.base import bootstrap_request
+    request = bootstrap_request(scheme='https')
     return request
 
 
 @pytest.fixture
 def context_stub():
     """
     Stub context object.
     """
     context = pyramid.testing.DummyResource()
     return context
 
 
 @pytest.fixture
 def config_stub(request, request_stub):
     """
     Set up pyramid.testing and return the Configurator.
     """
     config = pyramid.testing.setUp(request=request_stub)
     add_test_routes(config)
 
     @request.addfinalizer
     def cleanup():
         pyramid.testing.tearDown()
 
     return config
 
 
 @pytest.fixture
 def StubIntegrationType():
     class _StubIntegrationType(IntegrationTypeBase):
         """ Test integration type class """
 
         key = 'test'
         display_name = 'Test integration type'
         description = 'A test integration type for testing'
         icon = 'test_icon_html_image'
 
         def __init__(self, settings):
             super(_StubIntegrationType, self).__init__(settings)
             self.sent_events = []  # for testing
 
         def send_event(self, event):
             self.sent_events.append(event)
 
         def settings_schema(self):
             class SettingsSchema(colander.Schema):
                 test_string_field = colander.SchemaNode(
                     colander.String(),
                     missing=colander.required,
                     title='test string field',
                 )
                 test_int_field = colander.SchemaNode(
                     colander.Int(),
                     title='some integer setting',
                 )
             return SettingsSchema()
 
 
     integration_type_registry.register_integration_type(_StubIntegrationType)
     return _StubIntegrationType
 
 @pytest.fixture
 def stub_integration_settings():
     return {
         'test_string_field': 'some data',
         'test_int_field': 100,
     }
 
 
 @pytest.fixture
 def repo_integration_stub(request, repo_stub, StubIntegrationType,
         stub_integration_settings):
     integration = IntegrationModel().create(
         StubIntegrationType, settings=stub_integration_settings, enabled=True,
         name='test repo integration',
         repo=repo_stub, repo_group=None, child_repos_only=None)
 
     @request.addfinalizer
     def cleanup():
         IntegrationModel().delete(integration)
 
     return integration
 
 
 @pytest.fixture
 def repogroup_integration_stub(request, test_repo_group, StubIntegrationType,
     stub_integration_settings):
     integration = IntegrationModel().create(
         StubIntegrationType, settings=stub_integration_settings, enabled=True,
         name='test repogroup integration',
         repo=None, repo_group=test_repo_group, child_repos_only=True)
 
     @request.addfinalizer
     def cleanup():
         IntegrationModel().delete(integration)
 
     return integration
 
 
 @pytest.fixture
 def repogroup_recursive_integration_stub(request, test_repo_group,
     StubIntegrationType, stub_integration_settings):
     integration = IntegrationModel().create(
         StubIntegrationType, settings=stub_integration_settings, enabled=True,
         name='test recursive repogroup integration',
         repo=None, repo_group=test_repo_group, child_repos_only=False)
 
     @request.addfinalizer
     def cleanup():
         IntegrationModel().delete(integration)
 
     return integration
 
 
 @pytest.fixture
 def global_integration_stub(request, StubIntegrationType,
     stub_integration_settings):
     integration = IntegrationModel().create(
         StubIntegrationType, settings=stub_integration_settings, enabled=True,
         name='test global integration',
         repo=None, repo_group=None, child_repos_only=None)
 
     @request.addfinalizer
     def cleanup():
         IntegrationModel().delete(integration)
 
     return integration
 
 
 @pytest.fixture
 def root_repos_integration_stub(request, StubIntegrationType,
     stub_integration_settings):
     integration = IntegrationModel().create(
         StubIntegrationType, settings=stub_integration_settings, enabled=True,
         name='test global integration',
         repo=None, repo_group=None, child_repos_only=True)
 
     @request.addfinalizer
     def cleanup():
         IntegrationModel().delete(integration)
 
     return integration
 
 
 @pytest.fixture
 def local_dt_to_utc():
     def _factory(dt):
         return dt.replace(tzinfo=dateutil.tz.tzlocal()).astimezone(
             dateutil.tz.tzutc()).replace(tzinfo=None)
     return _factory
 
 
 @pytest.fixture
 def disable_anonymous_user(request, pylonsapp):
     set_anonymous_access(False)
 
     @request.addfinalizer
     def cleanup():
         set_anonymous_access(True)