vcs: moved svn proxy settings into vcs related settings

- vcs middleware now returns dummyHandler if not configured
- propagate the per-repo vcs related configs into VCSMiddleware
- moved the forms arround and group things into more logical blocks

rhodecode/controllers/admin/settings.py

@@ -1,793 +1,796 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  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/
 
 
 """
 settings controller for rhodecode admin
 """
 
 import collections
 import logging
 import urllib2
 
 import datetime
 import formencode
 from formencode import htmlfill
 import packaging.version
 from pylons import request, tmpl_context as c, url, config
 from pylons.controllers.util import redirect
 from pylons.i18n.translation import _, lazy_ugettext
 from webob.exc import HTTPBadRequest
 
 import rhodecode
 from rhodecode.admin.navigation import navigation_list
 from rhodecode.lib import auth
 from rhodecode.lib import helpers as h
 from rhodecode.lib.auth import LoginRequired, HasPermissionAllDecorator
 from rhodecode.lib.base import BaseController, render
 from rhodecode.lib.celerylib import tasks, run_task
 from rhodecode.lib.utils import repo2db_mapper
 from rhodecode.lib.utils2 import (
     str2bool, safe_unicode, AttributeDict, safe_int)
 from rhodecode.lib.compat import OrderedDict
 from rhodecode.lib.ext_json import json
 from rhodecode.lib.utils import jsonify
 
 from rhodecode.model.db import RhodeCodeUi, Repository
 from rhodecode.model.forms import ApplicationSettingsForm, \
     ApplicationUiSettingsForm, ApplicationVisualisationForm, \
     LabsSettingsForm, IssueTrackerPatternsForm
 
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.notification import EmailNotificationModel
 from rhodecode.model.meta import Session
 from rhodecode.model.settings import (
     IssueTrackerSettingsModel, VcsSettingsModel, SettingNotFound,
     SettingsModel)
 
 from rhodecode.model.supervisor import SupervisorModel, SUPERVISOR_MASTER
 
 
 log = logging.getLogger(__name__)
 
 
 class SettingsController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
     # To properly map this controller, ensure your config/routing.py
     # file has a resource setup:
     #     map.resource('setting', 'settings', controller='admin/settings',
     #         path_prefix='/admin', name_prefix='admin_')
 
     @LoginRequired()
     def __before__(self):
         super(SettingsController, self).__before__()
         c.labs_active = str2bool(
             rhodecode.CONFIG.get('labs_settings_active', 'true'))
         c.navlist = navigation_list(request)
 
     def _get_hg_ui_settings(self):
         ret = RhodeCodeUi.query().all()
 
         if not ret:
             raise Exception('Could not get application ui settings !')
         settings = {}
         for each in ret:
             k = each.ui_key
             v = each.ui_value
             if k == '/':
                 k = 'root_path'
 
             if k in ['push_ssl', 'publish']:
                 v = str2bool(v)
 
             if k.find('.') != -1:
                 k = k.replace('.', '_')
 
             if each.ui_section in ['hooks', 'extensions']:
                 v = each.ui_active
 
             settings[each.ui_section + '_' + k] = v
         return settings
 
     @HasPermissionAllDecorator('hg.admin')
     @auth.CSRFRequired()
     @jsonify
     def delete_svn_pattern(self):
         if not request.is_xhr:
             raise HTTPBadRequest()
 
         delete_pattern_id = request.POST.get('delete_svn_pattern')
         model = VcsSettingsModel()
         try:
             model.delete_global_svn_pattern(delete_pattern_id)
         except SettingNotFound:
             raise HTTPBadRequest()
 
         Session().commit()
         return True
 
     @HasPermissionAllDecorator('hg.admin')
     @auth.CSRFRequired()
     def settings_vcs_update(self):
         """POST /admin/settings: All items in the collection"""
         # url('admin_settings_vcs')
         c.active = 'vcs'
 
         model = VcsSettingsModel()
         c.svn_branch_patterns = model.get_global_svn_branch_patterns()
         c.svn_tag_patterns = model.get_global_svn_tag_patterns()
 
         application_form = ApplicationUiSettingsForm()()
+
         try:
             form_result = application_form.to_python(dict(request.POST))
         except formencode.Invalid as errors:
             h.flash(
                 _("Some form inputs contain invalid data."),
                 category='error')
             return htmlfill.render(
                 render('admin/settings/settings.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False
             )
 
         try:
-            model.update_global_ssl_setting(form_result['web_push_ssl'])
             if c.visual.allow_repo_location_change:
                 model.update_global_path_setting(
                     form_result['paths_root_path'])
+
+            model.update_global_ssl_setting(form_result['web_push_ssl'])
             model.update_global_hook_settings(form_result)
-            model.create_global_svn_settings(form_result)
+
+            model.create_or_update_global_svn_settings(form_result)
             model.create_or_update_global_hg_settings(form_result)
             model.create_or_update_global_pr_settings(form_result)
         except Exception:
             log.exception("Exception while updating settings")
             h.flash(_('Error occurred during updating '
                       'application settings'), category='error')
         else:
             Session().commit()
             h.flash(_('Updated VCS settings'), category='success')
             return redirect(url('admin_settings_vcs'))
 
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=self._form_defaults(),
             encoding="UTF-8",
             force_defaults=False)
 
     @HasPermissionAllDecorator('hg.admin')
     def settings_vcs(self):
         """GET /admin/settings: All items in the collection"""
         # url('admin_settings_vcs')
         c.active = 'vcs'
         model = VcsSettingsModel()
         c.svn_branch_patterns = model.get_global_svn_branch_patterns()
         c.svn_tag_patterns = model.get_global_svn_tag_patterns()
 
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=self._form_defaults(),
             encoding="UTF-8",
             force_defaults=False)
 
     @HasPermissionAllDecorator('hg.admin')
     @auth.CSRFRequired()
     def settings_mapping_update(self):
         """POST /admin/settings/mapping: All items in the collection"""
         # url('admin_settings_mapping')
         c.active = 'mapping'
         rm_obsolete = request.POST.get('destroy', False)
         invalidate_cache = request.POST.get('invalidate', False)
         log.debug(
             'rescanning repo location with destroy obsolete=%s', rm_obsolete)
 
         if invalidate_cache:
             log.debug('invalidating all repositories cache')
             for repo in Repository.get_all():
                 ScmModel().mark_for_invalidation(repo.repo_name, delete=True)
 
         filesystem_repos = ScmModel().repo_scan()
         added, removed = repo2db_mapper(filesystem_repos, rm_obsolete)
         _repr = lambda l: ', '.join(map(safe_unicode, l)) or '-'
         h.flash(_('Repositories successfully '
                   'rescanned added: %s ; removed: %s') %
                 (_repr(added), _repr(removed)),
                 category='success')
         return redirect(url('admin_settings_mapping'))
 
     @HasPermissionAllDecorator('hg.admin')
     def settings_mapping(self):
         """GET /admin/settings/mapping: All items in the collection"""
         # url('admin_settings_mapping')
         c.active = 'mapping'
 
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=self._form_defaults(),
             encoding="UTF-8",
             force_defaults=False)
 
     @HasPermissionAllDecorator('hg.admin')
     @auth.CSRFRequired()
     def settings_global_update(self):
         """POST /admin/settings/global: All items in the collection"""
         # url('admin_settings_global')
         c.active = 'global'
         application_form = ApplicationSettingsForm()()
         try:
             form_result = application_form.to_python(dict(request.POST))
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('admin/settings/settings.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
 
         try:
             settings = [
                 ('title', 'rhodecode_title'),
                 ('realm', 'rhodecode_realm'),
                 ('pre_code', 'rhodecode_pre_code'),
                 ('post_code', 'rhodecode_post_code'),
                 ('captcha_public_key', 'rhodecode_captcha_public_key'),
                 ('captcha_private_key', 'rhodecode_captcha_private_key'),
             ]
             for setting, form_key in settings:
                 sett = SettingsModel().create_or_update_setting(
                     setting, form_result[form_key])
                 Session().add(sett)
 
             Session().commit()
             SettingsModel().invalidate_settings_cache()
             h.flash(_('Updated application settings'), category='success')
         except Exception:
             log.exception("Exception while updating application settings")
             h.flash(
                 _('Error occurred during updating application settings'),
                 category='error')
 
         return redirect(url('admin_settings_global'))
 
     @HasPermissionAllDecorator('hg.admin')
     def settings_global(self):
         """GET /admin/settings/global: All items in the collection"""
         # url('admin_settings_global')
         c.active = 'global'
 
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=self._form_defaults(),
             encoding="UTF-8",
             force_defaults=False)
 
     @HasPermissionAllDecorator('hg.admin')
     @auth.CSRFRequired()
     def settings_visual_update(self):
         """POST /admin/settings/visual: All items in the collection"""
         # url('admin_settings_visual')
         c.active = 'visual'
         application_form = ApplicationVisualisationForm()()
         try:
             form_result = application_form.to_python(dict(request.POST))
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('admin/settings/settings.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False
             )
 
         try:
             settings = [
                 ('show_public_icon', 'rhodecode_show_public_icon', 'bool'),
                 ('show_private_icon', 'rhodecode_show_private_icon', 'bool'),
                 ('stylify_metatags', 'rhodecode_stylify_metatags', 'bool'),
                 ('repository_fields', 'rhodecode_repository_fields', 'bool'),
                 ('dashboard_items', 'rhodecode_dashboard_items', 'int'),
                 ('admin_grid_items', 'rhodecode_admin_grid_items', 'int'),
                 ('show_version', 'rhodecode_show_version', 'bool'),
                 ('use_gravatar', 'rhodecode_use_gravatar', 'bool'),
                 ('markup_renderer', 'rhodecode_markup_renderer', 'unicode'),
                 ('gravatar_url', 'rhodecode_gravatar_url', 'unicode'),
                 ('clone_uri_tmpl', 'rhodecode_clone_uri_tmpl', 'unicode'),
                 ('support_url', 'rhodecode_support_url', 'unicode'),
                 ('show_revision_number', 'rhodecode_show_revision_number', 'bool'),
                 ('show_sha_length', 'rhodecode_show_sha_length', 'int'),
             ]
             for setting, form_key, type_ in settings:
                 sett = SettingsModel().create_or_update_setting(
                     setting, form_result[form_key], type_)
                 Session().add(sett)
 
             Session().commit()
             SettingsModel().invalidate_settings_cache()
             h.flash(_('Updated visualisation settings'), category='success')
         except Exception:
             log.exception("Exception updating visualization settings")
             h.flash(_('Error occurred during updating '
                       'visualisation settings'),
                     category='error')
 
         return redirect(url('admin_settings_visual'))
 
     @HasPermissionAllDecorator('hg.admin')
     def settings_visual(self):
         """GET /admin/settings/visual: All items in the collection"""
         # url('admin_settings_visual')
         c.active = 'visual'
 
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=self._form_defaults(),
             encoding="UTF-8",
             force_defaults=False)
 
     @HasPermissionAllDecorator('hg.admin')
     @auth.CSRFRequired()
     def settings_issuetracker_test(self):
         if request.is_xhr:
             return h.urlify_commit_message(
                 request.POST.get('test_text', ''),
                 'repo_group/test_repo1')
         else:
             raise HTTPBadRequest()
 
     @HasPermissionAllDecorator('hg.admin')
     @auth.CSRFRequired()
     def settings_issuetracker_delete(self):
         uid = request.POST.get('uid')
         IssueTrackerSettingsModel().delete_entries(uid)
         h.flash(_('Removed issue tracker entry'), category='success')
         return redirect(url('admin_settings_issuetracker'))
 
     @HasPermissionAllDecorator('hg.admin')
     def settings_issuetracker(self):
         """GET /admin/settings/issue-tracker: All items in the collection"""
         # url('admin_settings_issuetracker')
         c.active = 'issuetracker'
         defaults = SettingsModel().get_all_settings()
 
         entry_key = 'rhodecode_issuetracker_pat_'
 
         c.issuetracker_entries = {}
         for k, v in defaults.items():
             if k.startswith(entry_key):
                 uid = k[len(entry_key):]
                 c.issuetracker_entries[uid] = None
 
         for uid in c.issuetracker_entries:
             c.issuetracker_entries[uid] = AttributeDict({
                 'pat': defaults.get('rhodecode_issuetracker_pat_' + uid),
                 'url': defaults.get('rhodecode_issuetracker_url_' + uid),
                 'pref': defaults.get('rhodecode_issuetracker_pref_' + uid),
                 'desc': defaults.get('rhodecode_issuetracker_desc_' + uid),
             })
 
         return render('admin/settings/settings.html')
 
     @HasPermissionAllDecorator('hg.admin')
     @auth.CSRFRequired()
     def settings_issuetracker_save(self):
         settings_model = IssueTrackerSettingsModel()
 
         form = IssueTrackerPatternsForm()().to_python(request.POST)
         for uid in form['delete_patterns']:
             settings_model.delete_entries(uid)
 
         for pattern in form['patterns']:
             for setting, value, type_ in pattern:
                 sett = settings_model.create_or_update_setting(
                     setting, value, type_)
                 Session().add(sett)
 
             Session().commit()
 
         SettingsModel().invalidate_settings_cache()
         h.flash(_('Updated issue tracker entries'), category='success')
         return redirect(url('admin_settings_issuetracker'))
 
     @HasPermissionAllDecorator('hg.admin')
     @auth.CSRFRequired()
     def settings_email_update(self):
         """POST /admin/settings/email: All items in the collection"""
         # url('admin_settings_email')
         c.active = 'email'
 
         test_email = request.POST.get('test_email')
 
         if not test_email:
             h.flash(_('Please enter email address'), category='error')
             return redirect(url('admin_settings_email'))
 
         email_kwargs = {
             'date': datetime.datetime.now(),
             'user': c.rhodecode_user,
             'rhodecode_version': c.rhodecode_version
         }
 
         (subject, headers, email_body,
          email_body_plaintext) = EmailNotificationModel().render_email(
             EmailNotificationModel.TYPE_EMAIL_TEST, **email_kwargs)
 
         recipients = [test_email] if test_email else None
 
         run_task(tasks.send_email, recipients, subject,
                  email_body_plaintext, email_body)
 
         h.flash(_('Send email task created'), category='success')
         return redirect(url('admin_settings_email'))
 
     @HasPermissionAllDecorator('hg.admin')
     def settings_email(self):
         """GET /admin/settings/email: All items in the collection"""
         # url('admin_settings_email')
         c.active = 'email'
         c.rhodecode_ini = rhodecode.CONFIG
 
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=self._form_defaults(),
             encoding="UTF-8",
             force_defaults=False)
 
     @HasPermissionAllDecorator('hg.admin')
     @auth.CSRFRequired()
     def settings_hooks_update(self):
         """POST or DELETE /admin/settings/hooks: All items in the collection"""
         # url('admin_settings_hooks')
         c.active = 'hooks'
         if c.visual.allow_custom_hooks_settings:
             ui_key = request.POST.get('new_hook_ui_key')
             ui_value = request.POST.get('new_hook_ui_value')
 
             hook_id = request.POST.get('hook_id')
             new_hook = False
 
             model = SettingsModel()
             try:
                 if ui_value and ui_key:
                     model.create_or_update_hook(ui_key, ui_value)
                     h.flash(_('Added new hook'), category='success')
                     new_hook = True
                 elif hook_id:
                     RhodeCodeUi.delete(hook_id)
                     Session().commit()
 
                 # check for edits
                 update = False
                 _d = request.POST.dict_of_lists()
                 for k, v in zip(_d.get('hook_ui_key', []),
                                 _d.get('hook_ui_value_new', [])):
                     model.create_or_update_hook(k, v)
                     update = True
 
                 if update and not new_hook:
                     h.flash(_('Updated hooks'), category='success')
                 Session().commit()
             except Exception:
                 log.exception("Exception during hook creation")
                 h.flash(_('Error occurred during hook creation'),
                         category='error')
 
         return redirect(url('admin_settings_hooks'))
 
     @HasPermissionAllDecorator('hg.admin')
     def settings_hooks(self):
         """GET /admin/settings/hooks: All items in the collection"""
         # url('admin_settings_hooks')
         c.active = 'hooks'
 
         model = SettingsModel()
         c.hooks = model.get_builtin_hooks()
         c.custom_hooks = model.get_custom_hooks()
 
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=self._form_defaults(),
             encoding="UTF-8",
             force_defaults=False)
 
     @HasPermissionAllDecorator('hg.admin')
     def settings_search(self):
         """GET /admin/settings/search: All items in the collection"""
         # url('admin_settings_search')
         c.active = 'search'
 
         from rhodecode.lib.index import searcher_from_config
         searcher = searcher_from_config(config)
         c.statistics = searcher.statistics()
 
         return render('admin/settings/settings.html')
 
     @HasPermissionAllDecorator('hg.admin')
     def settings_system(self):
         """GET /admin/settings/system: All items in the collection"""
         # url('admin_settings_system')
         snapshot = str2bool(request.GET.get('snapshot'))
         c.active = 'system'
 
         defaults = self._form_defaults()
         c.rhodecode_ini = rhodecode.CONFIG
         c.rhodecode_update_url = defaults.get('rhodecode_update_url')
         server_info = ScmModel().get_server_info(request.environ)
         for key, val in server_info.iteritems():
             setattr(c, key, val)
 
         if c.disk['percent'] > 90:
             h.flash(h.literal(_(
                 'Critical: your disk space is very low <b>%s%%</b> used' %
                 c.disk['percent'])), 'error')
         elif c.disk['percent'] > 70:
             h.flash(h.literal(_(
                 'Warning: your disk space is running low <b>%s%%</b> used' %
                 c.disk['percent'])), 'warning')
 
         try:
             c.uptime_age = h._age(
                 h.time_to_datetime(c.boot_time), False, show_suffix=False)
         except TypeError:
             c.uptime_age = c.boot_time
 
         try:
             c.system_memory = '%s/%s, %s%% (%s%%) used%s' % (
                 h.format_byte_size_binary(c.memory['used']),
                 h.format_byte_size_binary(c.memory['total']),
                 c.memory['percent2'],
                 c.memory['percent'],
                 ' %s' % c.memory['error'] if 'error' in c.memory else '')
         except TypeError:
             c.system_memory = 'NOT AVAILABLE'
 
         rhodecode_ini_safe = rhodecode.CONFIG.copy()
         blacklist = [
             'rhodecode_license_key',
             'routes.map',
             'pylons.h',
             'pylons.app_globals',
             'pylons.environ_config',
             'sqlalchemy.db1.url',
             ('app_conf', 'sqlalchemy.db1.url')
         ]
         for k in blacklist:
             if isinstance(k, tuple):
                 section, key = k
                 if section in rhodecode_ini_safe:
                     rhodecode_ini_safe[section].pop(key, None)
             else:
                 rhodecode_ini_safe.pop(k, None)
 
         c.rhodecode_ini_safe = rhodecode_ini_safe
 
         # TODO: marcink, figure out how to allow only selected users to do this
         c.allowed_to_snapshot = False
 
         if snapshot:
             if c.allowed_to_snapshot:
                 return render('admin/settings/settings_system_snapshot.html')
             else:
                 h.flash('You are not allowed to do this', category='warning')
 
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
 
     @staticmethod
     def get_update_data(update_url):
         """Return the JSON update data."""
         ver = rhodecode.__version__
         log.debug('Checking for upgrade on `%s` server', update_url)
         opener = urllib2.build_opener()
         opener.addheaders = [('User-agent', 'RhodeCode-SCM/%s' % ver)]
         response = opener.open(update_url)
         response_data = response.read()
         data = json.loads(response_data)
 
         return data
 
     @HasPermissionAllDecorator('hg.admin')
     def settings_system_update(self):
         """GET /admin/settings/system/updates: All items in the collection"""
         # url('admin_settings_system_update')
         defaults = self._form_defaults()
         update_url = defaults.get('rhodecode_update_url', '')
 
         _err = lambda s: '<div style="color:#ff8888; padding:4px 0px">%s</div>' % (s)
         try:
             data = self.get_update_data(update_url)
         except urllib2.URLError as e:
             log.exception("Exception contacting upgrade server")
             return _err('Failed to contact upgrade server: %r' % e)
         except ValueError as e:
             log.exception("Bad data sent from update server")
             return _err('Bad data sent from update server')
 
         latest = data['versions'][0]
 
         c.update_url = update_url
         c.latest_data = latest
         c.latest_ver = latest['version']
         c.cur_ver = rhodecode.__version__
         c.should_upgrade = False
 
         if (packaging.version.Version(c.latest_ver) >
                 packaging.version.Version(c.cur_ver)):
             c.should_upgrade = True
         c.important_notices = latest['general']
 
         return render('admin/settings/settings_system_update.html')
 
     @HasPermissionAllDecorator('hg.admin')
     def settings_supervisor(self):
         c.rhodecode_ini = rhodecode.CONFIG
         c.active = 'supervisor'
 
         c.supervisor_procs = OrderedDict([
             (SUPERVISOR_MASTER, {}),
         ])
 
         c.log_size = 10240
         supervisor = SupervisorModel()
 
         _connection = supervisor.get_connection(
             c.rhodecode_ini.get('supervisor.uri'))
         c.connection_error = None
         try:
             _connection.supervisor.getAllProcessInfo()
         except Exception as e:
             c.connection_error = str(e)
             log.exception("Exception reading supervisor data")
             return render('admin/settings/settings.html')
 
         groupid = c.rhodecode_ini.get('supervisor.group_id')
 
         # feed our group processes to the main
         for proc in supervisor.get_group_processes(_connection, groupid):
             c.supervisor_procs[proc['name']] = {}
 
         for k in c.supervisor_procs.keys():
             try:
                 # master process info
                 if k == SUPERVISOR_MASTER:
                     _data = supervisor.get_master_state(_connection)
                     _data['name'] = 'supervisor master'
                     _data['description'] = 'pid %s, id: %s, ver: %s' % (
                         _data['pid'], _data['id'], _data['ver'])
                     c.supervisor_procs[k] = _data
                 else:
                     procid = groupid + ":" + k
                     c.supervisor_procs[k] = supervisor.get_process_info(_connection, procid)
             except Exception as e:
                 log.exception("Exception reading supervisor data")
                 c.supervisor_procs[k] = {'_rhodecode_error': str(e)}
 
         return render('admin/settings/settings.html')
 
     @HasPermissionAllDecorator('hg.admin')
     def settings_supervisor_log(self, procid):
         import rhodecode
         c.rhodecode_ini = rhodecode.CONFIG
         c.active = 'supervisor_tail'
 
         supervisor = SupervisorModel()
         _connection = supervisor.get_connection(c.rhodecode_ini.get('supervisor.uri'))
         groupid = c.rhodecode_ini.get('supervisor.group_id')
         procid = groupid + ":" + procid if procid != SUPERVISOR_MASTER else procid
 
         c.log_size = 10240
         offset = abs(safe_int(request.GET.get('offset', c.log_size))) * -1
         c.log = supervisor.read_process_log(_connection, procid, offset, 0)
 
         return render('admin/settings/settings.html')
 
     @HasPermissionAllDecorator('hg.admin')
     @auth.CSRFRequired()
     def settings_labs_update(self):
         """POST /admin/settings/labs: All items in the collection"""
         # url('admin_settings/labs', method={'POST'})
         c.active = 'labs'
 
         application_form = LabsSettingsForm()()
         try:
             form_result = application_form.to_python(dict(request.POST))
         except formencode.Invalid as errors:
             h.flash(
                 _('Some form inputs contain invalid data.'),
                 category='error')
             return htmlfill.render(
                 render('admin/settings/settings.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding='UTF-8',
                 force_defaults=False
             )
 
         try:
             session = Session()
             for setting in _LAB_SETTINGS:
                 setting_name = setting.key[len('rhodecode_'):]
                 sett = SettingsModel().create_or_update_setting(
                     setting_name, form_result[setting.key], setting.type)
                 session.add(sett)
 
         except Exception:
             log.exception('Exception while updating lab settings')
             h.flash(_('Error occurred during updating labs settings'),
                     category='error')
         else:
             Session().commit()
             SettingsModel().invalidate_settings_cache()
             h.flash(_('Updated Labs settings'), category='success')
             return redirect(url('admin_settings_labs'))
 
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=self._form_defaults(),
             encoding='UTF-8',
             force_defaults=False)
 
     @HasPermissionAllDecorator('hg.admin')
     def settings_labs(self):
         """GET /admin/settings/labs: All items in the collection"""
         # url('admin_settings_labs')
         if not c.labs_active:
             redirect(url('admin_settings'))
 
         c.active = 'labs'
         c.lab_settings = _LAB_SETTINGS
 
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=self._form_defaults(),
             encoding='UTF-8',
             force_defaults=False)
 
     def _form_defaults(self):
         defaults = SettingsModel().get_all_settings()
         defaults.update(self._get_hg_ui_settings())
         defaults.update({
             'new_svn_branch': '',
             'new_svn_tag': '',
         })
         return defaults
 
 
 # :param key: name of the setting including the 'rhodecode_' prefix
 # :param type: the RhodeCodeSetting type to use.
 # :param group: the i18ned group in which we should dispaly this setting
 # :param label: the i18ned label we should display for this setting
 # :param help: the i18ned help we should dispaly for this setting
 LabSetting = collections.namedtuple(
     'LabSetting', ('key', 'type', 'group', 'label', 'help'))
 
 
 # This list has to be kept in sync with the form
 # rhodecode.model.forms.LabsSettingsForm.
 _LAB_SETTINGS = [
 
 ]

rhodecode/lib/base.py

@@ -1,587 +1,587 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  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 ipaddress
 
 from paste.auth.basic import AuthBasicAuthenticator
 from paste.httpexceptions import HTTPUnauthorized, HTTPForbidden, get_exception
 from paste.httpheaders import WWW_AUTHENTICATE, AUTHORIZATION
 from pylons import config, tmpl_context as c, request, session, url
 from pylons.controllers import WSGIController
 from pylons.controllers.util import redirect
 from pylons.i18n import translation
 # marcink: don't remove this import
 from pylons.templating import render_mako as render  # noqa
 from pylons.i18n.translation import _
 from webob.exc import HTTPFound
 
 
 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.lib.vcs.exceptions import RepositoryRequirementError
 from rhodecode.model import meta
 from rhodecode.model.db import Repository, User
 from rhodecode.model.notification import NotificationModel
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.settings import VcsSettingsModel, SettingsModel
 
 
 log = logging.getLogger(__name__)
 
 
 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(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 vcs_operation_context(
         environ, repo_name, username, action, scm, check_locking=True):
     """
     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),
         'hooks': get_enabled_hook_classes(ui_settings),
     }
     return extras
 
 
 class BasicAuth(AuthBasicAuthenticator):
 
     def __init__(self, realm, authfunc, registry, auth_http_code=None,
                  initial_call_detection=False):
         self.realm = realm
         self.initial_call = initial_call_detection
         self.authfunc = authfunc
         self.registry = registry
         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):
                 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 attach_context_attributes(context, request):
     """
     Attach variables into template context called `c`, please note that
     request could be pylons or pyramid request in here.
     """
     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 = md5(
         config.get('beaker.session.secret', '') +
         rhodecode.__version__)[:8]
 
     # Default language set for the incoming request
     context.language = translation.get_lang()[0]
 
     # Visual options
     context.visual = AttributeDict({})
 
-    # DB store
+    # 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.rhodecode_support_url = \
         rc_config.get('rhodecode_support_url') or url('rhodecode_support')
 
     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')
 
     # 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
         },
         'pylons_dispatch': {
             # 'controller': request.environ['pylons.routes_dict']['controller'],
             # 'action': request.environ['pylons.routes_dict']['action'],
         },
         '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
 
     context.csrf_token = auth.get_csrf_token()
     context.backends = rhodecode.BACKENDS.keys()
     context.backends.sort()
     context.unread_notifications = NotificationModel().get_unread_cnt_for_user(
         context.rhodecode_user.user_id)
 
 
 def get_auth_user(environ):
     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 are sure user exists.
         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.
         set_rhodecode_config(config)
         attach_context_attributes(c, request)
 
         # 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)
 
         default_lang = c.language
         user_lang = c.language
         try:
             user_obj = self._rhodecode_user.get_instance()
             if user_obj:
                 user_lang = user_obj.user_data.get('language')
         except Exception:
             log.exception('Failed to fetch user language for user %s',
                           self._rhodecode_user)
 
         if user_lang and user_lang != default_lang:
             log.debug('set language to %s for user %s', user_lang,
                       self._rhodecode_user)
             translation.set_lang(user_lang)
 
     def _dispatch_redirect(self, with_url, environ, start_response):
         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
 
         # 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
         c.rhodecode_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)
         )
 
         # TODO: Maybe this should be move to pyramid to cover all views.
         # check user attributes for password change flag
         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)
 
             skip_user_check_urls = [
                 'error.document', 'login.logout', 'login.index',
                 'admin/my_account.my_account_password',
                 'admin/my_account.my_account_password_update'
             ]
             if _route_name not in skip_user_check_urls:
                 return self._dispatch_redirect(
                     url('my_account_password'), environ, start_response)
 
         return WSGIController.__call__(self, environ, start_response)
 
 
 class BaseRepoController(BaseController):
     """
     Base class for controllers responsible for loading all needed data for
     repository loaded items are
 
     c.rhodecode_repo: instance of scm repository
     c.rhodecode_db_repo: instance of db
     c.repository_requirements_missing: shows that repository specific data
         could not be displayed due to the missing requirements
     c.repository_pull_requests: show number of open pull requests
     """
 
     def __before__(self):
         super(BaseRepoController, self).__before__()
         if c.repo_name:  # extracted from routes
             db_repo = Repository.get_by_repo_name(c.repo_name)
             if not db_repo:
                 return
 
             log.debug(
                 'Found repository in database %s with state `%s`',
                 safe_unicode(db_repo), safe_unicode(db_repo.repo_state))
             route = getattr(request.environ.get('routes.route'), 'name', '')
 
             # allow to delete repos that are somehow damages in filesystem
             if route in ['delete_repo']:
                 return
 
             if db_repo.repo_state in [Repository.STATE_PENDING]:
                 if route in ['repo_creating_home']:
                     return
                 check_url = url('repo_creating_home', repo_name=c.repo_name)
                 return redirect(check_url)
 
             self.rhodecode_db_repo = db_repo
 
             missing_requirements = False
             try:
                 self.rhodecode_repo = self.rhodecode_db_repo.scm_instance()
             except RepositoryRequirementError as e:
                 missing_requirements = True
                 self._handle_missing_requirements(e)
 
             if self.rhodecode_repo is None and not missing_requirements:
                 log.error('%s this repository is present in database but it '
                           'cannot be created as an scm instance', c.repo_name)
 
                 h.flash(_(
                     "The repository at %(repo_name)s cannot be located.") %
                     {'repo_name': c.repo_name},
                     category='error', ignore_duplicate=True)
                 redirect(url('home'))
 
             # update last change according to VCS data
             if not missing_requirements:
                 commit = db_repo.get_commit(
                     pre_load=["author", "date", "message", "parents"])
                 db_repo.update_commit_cache(commit)
 
             # Prepare context
             c.rhodecode_db_repo = db_repo
             c.rhodecode_repo = self.rhodecode_repo
             c.repository_requirements_missing = missing_requirements
 
             self._update_global_counters(self.scm_model, db_repo)
 
     def _update_global_counters(self, scm_model, db_repo):
         """
         Base variables that are exposed to every page of repository
         """
         c.repository_pull_requests = scm_model.get_pull_requests(db_repo)
 
     def _handle_missing_requirements(self, error):
         self.rhodecode_repo = None
         log.error(
             'Requirements are missing for repository %s: %s',
             c.repo_name, error.message)
 
         summary_url = url('summary_home', repo_name=c.repo_name)
         statistics_url = url('edit_repo_statistics', repo_name=c.repo_name)
         settings_update_url = url('repo', repo_name=c.repo_name)
         path = request.path
         should_redirect = (
             path not in (summary_url, settings_update_url)
             and '/settings' not in path or path == statistics_url
         )
         if should_redirect:
             redirect(summary_url)

rhodecode/lib/helpers.py

@@ -1,1966 +1,1966 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  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
 import pygments
 
 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 pylons import url as pylons_url
 from pylons.i18n.translation import _, ungettext
 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.annotate import annotate_highlight
 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
 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):
     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.
     """
     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 asset(path, ver=None):
     """
     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 = {}
     if ver:
         query = {'ver': ver}
     return request.static_path(
         'rhodecode:public/{}'.format(path), _query=query)
 
 
 def html_escape(text, html_escape_table=None):
     """Produce entities within text."""
     if not html_escape_table:
         html_escape_table = {
             "&": "&amp;",
             '"': "&quot;",
             "'": "&apos;",
             ">": "&gt;",
             "<": "&lt;",
         }
     return "".join(html_escape_table.get(c, c) for c in text)
 
 
 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,
                 url('files_home',
                     repo_name=repo_name,
                     revision=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,
                     url('files_home',
                         repo_name=repo_name,
                         revision=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 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 pygmentize(filenode, **kwargs):
     """
     pygmentize function using pygments
 
     :param filenode:
     """
     lexer = get_custom_lexer(filenode.extension) or filenode.lexer
     return literal(code_highlight(filenode.content, lexer,
                                   CodeHtmlFormatter(**kwargs)))
 
 
 def pygmentize_annotation(repo_name, filenode, **kwargs):
     """
     pygmentize function for annotation
 
     :param filenode:
     """
 
     color_dict = {}
 
     def gen_color(n=10000):
         """generator for getting n of evenly distributed colors using
         hsv color and golden ratio. It always return same order of colors
 
         :returns: RGB tuple
         """
 
         def hsv_to_rgb(h, s, v):
             if s == 0.0:
                 return v, v, v
             i = int(h * 6.0)  # XXX assume int() truncates!
             f = (h * 6.0) - i
             p = v * (1.0 - s)
             q = v * (1.0 - s * f)
             t = v * (1.0 - s * (1.0 - f))
             i = i % 6
             if i == 0:
                 return v, t, p
             if i == 1:
                 return q, v, p
             if i == 2:
                 return p, v, t
             if i == 3:
                 return p, q, v
             if i == 4:
                 return t, p, v
             if i == 5:
                 return v, p, q
 
         golden_ratio = 0.618033988749895
         h = 0.22717784590367374
 
         for _ in xrange(n):
             h += golden_ratio
             h %= 1
             HSV_tuple = [h, 0.95, 0.95]
             RGB_tuple = hsv_to_rgb(*HSV_tuple)
             yield map(lambda x: str(int(x * 256)), RGB_tuple)
 
     cgenerator = gen_color()
 
     def get_color_string(commit_id):
         if commit_id in color_dict:
             col = color_dict[commit_id]
         else:
             col = color_dict[commit_id] = cgenerator.next()
         return "color: rgb(%s)! important;" % (', '.join(col))
 
     def url_func(repo_name):
 
         def _url_func(commit):
             author = commit.author
             date = commit.date
             message = tooltip(commit.message)
 
             tooltip_html = ("<div style='font-size:0.8em'><b>Author:</b>"
                             " %s<br/><b>Date:</b> %s</b><br/><b>Message:"
                             "</b> %s<br/></div>")
 
             tooltip_html = tooltip_html % (author, date, message)
             lnk_format = '%5s:%s' % ('r%s' % commit.idx, commit.short_id)
             uri = link_to(
                 lnk_format,
                 url('changeset_home', repo_name=repo_name,
                     revision=commit.raw_id),
                 style=get_color_string(commit.raw_id),
                 class_='tooltip',
                 title=tooltip_html
             )
 
             uri += '\n'
             return uri
         return _url_func
 
     return literal(annotate_highlight(filenode, url_func(repo_name), **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):
 
     def pop_messages(self):
         """Return all accumulated messages and delete them from the session.
 
         The return value is a list of ``Message`` objects.
         """
         from pylons import session
 
         messages = []
 
         # 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
 
 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)
 
     # detect if we have a timezone info, otherwise, add it
     if isinstance(datetime_iso, datetime) and not datetime_iso.tzinfo:
         tzinfo = '+00:00'
 
         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}" 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):
-    from rhodecode import CONFIG
     if is_svn(repository):
-        if not CONFIG.get('rhodecode_proxy_subversion_http_requests', False):
-            return True
+        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),
             url('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.firstname or user.lastname:
                 return '%s %s &lt;%s&gt;' % (user.firstname, user.lastname, 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(author, show_disabled=False):
     from rhodecode.lib.utils import PartialRenderer
     _render = PartialRenderer('base/base.html')
     return _render('gravatar_with_user', author, show_disabled=show_disabled)
 
 
 def desc_stylize(value):
     """
     converts tags from value into html equivalent
 
     :param value:
     """
     if not value:
         return ''
 
     value = re.sub(r'\[see\ \=\>\ *([a-zA-Z0-9\/\=\?\&\ \:\/\.\-]*)\]',
                    '<div class="metatag" tag="see">see =&gt; \\1 </div>', value)
     value = re.sub(r'\[license\ \=\>\ *([a-zA-Z0-9\/\=\?\&\ \:\/\.\-]*)\]',
                    '<div class="metatag" tag="license"><a href="http:\/\/www.opensource.org/licenses/\\1">\\1</a></div>', value)
     value = re.sub(r'\[(requires|recommends|conflicts|base)\ \=\>\ *([a-zA-Z0-9\-\/]*)\]',
                    '<div class="metatag" tag="\\1">\\1 =&gt; <a href="/\\2">\\2</a></div>', value)
     value = re.sub(r'\[(lang|language)\ \=\>\ *([a-zA-Z\-\/\#\+]*)\]',
                    '<div class="metatag" tag="lang">\\2</div>', value)
     value = re.sub(r'\[([a-z]+)\]',
                    '<div class="metatag" tag="\\1">\\1</div>', value)
 
     return value
 
 
 def escaped_stylize(value):
     """
     converts tags from value into html equivalent, but escaping its value first
     """
     if not value:
         return ''
 
     # Using default webhelper escape method, but has to force it as a
     # plain unicode instead of a markup tag to be used in regex expressions
     value = unicode(escape(safe_unicode(value)))
 
     value = re.sub(r'\[see\ \=\&gt;\ *([a-zA-Z0-9\/\=\?\&amp;\ \:\/\.\-]*)\]',
                    '<div class="metatag" tag="see">see =&gt; \\1 </div>', value)
     value = re.sub(r'\[license\ \=\&gt;\ *([a-zA-Z0-9\/\=\?\&amp;\ \:\/\.\-]*)\]',
                    '<div class="metatag" tag="license"><a href="http:\/\/www.opensource.org/licenses/\\1">\\1</a></div>', value)
     value = re.sub(r'\[(requires|recommends|conflicts|base)\ \=\&gt;\ *([a-zA-Z0-9\-\/]*)\]',
                    '<div class="metatag" tag="\\1">\\1 =&gt; <a href="/\\2">\\2</a></div>', value)
     value = re.sub(r'\[(lang|language)\ \=\&gt;\ *([a-zA-Z\-\/\#\+]*)\]',
                    '<div class="metatag" tag="lang">\\2</div>', value)
     value = re.sub(r'\[([a-z]+)\]',
                    '<div class="metatag" tag="\\1">\\1</div>', value)
 
     return 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 'firstname.lastname' syntax
         _dot_split = prefix.rsplit('.', 1)
         if len(_dot_split) == 2:
             initials = [_dot_split[0][0], _dot_split[1][0]]
         else:
             initials = [prefix[0], server[0]]
 
         # then try to replace either firtname or lastname
         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):
     # 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 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 changed_tooltip(nodes):
     """
     Generates a html string for changed nodes in commit page.
     It limits the output to 30 entries
 
     :param nodes: LazyNodesGenerator
     """
     if nodes:
         pref = ': <br/> '
         suf = ''
         if len(nodes) > 30:
             suf = '<br/>' + _(' and %s more') % (len(nodes) - 30)
         return literal(pref + '<br/> '.join([safe_unicode(x.path)
                                              for x in nodes[:30]]) + suf)
     else:
         return ': ' + _('No Files')
 
 
 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, url('repo_group_home', group_name=group.group_name))
         for group in repo.groups_with_parents
     ] + [
         link_to(repo.just_name, url('summary_home', repo_name=repo.repo_name))
     ]
 
     return literal(' &raquo; '.join(path))
 
 
 def format_byte_size_binary(file_size):
     """
     Formats file/folder sizes to standard.
     """
     formatted_size = format_byte_size(file_size, binary=True)
     return formatted_size
 
 
 def fancy_file_stats(stats):
     """
     Displays a fancy two colored bar for number of added/deleted
     lines of code on file
 
     :param stats: two element list of added/deleted lines of code
     """
     from rhodecode.lib.diffs import NEW_FILENODE, DEL_FILENODE, \
         MOD_FILENODE, RENAMED_FILENODE, CHMOD_FILENODE, BIN_FILENODE
 
     def cgen(l_type, a_v, d_v):
         mapping = {'tr': 'top-right-rounded-corner-mid',
                    'tl': 'top-left-rounded-corner-mid',
                    'br': 'bottom-right-rounded-corner-mid',
                    'bl': 'bottom-left-rounded-corner-mid'}
         map_getter = lambda x: mapping[x]
 
         if l_type == 'a' and d_v:
             #case when added and deleted are present
             return ' '.join(map(map_getter, ['tl', 'bl']))
 
         if l_type == 'a' and not d_v:
             return ' '.join(map(map_getter, ['tr', 'br', 'tl', 'bl']))
 
         if l_type == 'd' and a_v:
             return ' '.join(map(map_getter, ['tr', 'br']))
 
         if l_type == 'd' and not a_v:
             return ' '.join(map(map_getter, ['tr', 'br', 'tl', 'bl']))
 
     a, d = stats['added'], stats['deleted']
     width = 100
 
     if stats['binary']:  # binary operations like chmod/rename etc
         lbl = []
         bin_op = 0  # undefined
 
         # prefix with bin for binary files
         if BIN_FILENODE in stats['ops']:
             lbl += ['bin']
 
         if NEW_FILENODE in stats['ops']:
             lbl += [_('new file')]
             bin_op = NEW_FILENODE
         elif MOD_FILENODE in stats['ops']:
             lbl += [_('mod')]
             bin_op = MOD_FILENODE
         elif DEL_FILENODE in stats['ops']:
             lbl += [_('del')]
             bin_op = DEL_FILENODE
         elif RENAMED_FILENODE in stats['ops']:
             lbl += [_('rename')]
             bin_op = RENAMED_FILENODE
 
         # chmod can go with other operations, so we add a + to lbl if needed
         if CHMOD_FILENODE in stats['ops']:
             lbl += [_('chmod')]
             if bin_op == 0:
                 bin_op = CHMOD_FILENODE
 
         lbl = '+'.join(lbl)
         b_a = '<div class="bin bin%s %s" style="width:100%%">%s</div>' \
               % (bin_op, cgen('a', a_v='', d_v=0), lbl)
         b_d = '<div class="bin bin1" style="width:0%%"></div>'
         return literal('<div style="width:%spx">%s%s</div>' % (width, b_a, b_d))
 
     t = stats['added'] + stats['deleted']
     unit = float(width) / (t or 1)
 
     # needs > 9% of width to be visible or 0 to be hidden
     a_p = max(9, unit * a) if a > 0 else 0
     d_p = max(9, unit * d) if d > 0 else 0
     p_sum = a_p + d_p
 
     if p_sum > width:
         #adjust the percentage to be == 100% since we adjusted to 9
         if a_p > d_p:
             a_p = a_p - (p_sum - width)
         else:
             d_p = d_p - (p_sum - width)
 
     a_v = a if a > 0 else ''
     d_v = d if d > 0 else ''
 
     d_a = '<div class="added %s" style="width:%s%%">%s</div>' % (
         cgen('a', a_v, d_v), a_p, a_v
     )
     d_d = '<div class="deleted %s" style="width:%s%%">%s</div>' % (
         cgen('d', a_v, d_v), d_p, d_v
     )
     return literal('<div style="width:%spx">%s%s</div>' % (width, d_a, d_d))
 
 
 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
     """
     from pylons import url  # doh, we need to re-import url to mock it later
     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': url('changeset_home', repo_name=repository,
                        revision=commit_id, qualified=True),
             '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):
     pref = ''
     if match_obj.group().startswith(' '):
         pref = ' '
 
     issue_id = ''.join(match_obj.groups())
     tmpl = (
         '%(pref)s<a class="%(cls)s" href="%(url)s">'
         '%(issue-prefix)s%(id-repr)s'
         '</a>')
 
     (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, config=None):
     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)
 
         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 rst(source, mentions=False):
     return literal('<div class="rst-block">%s</div>' %
                    MarkupRenderer.rst(source, mentions=mentions))
 
 
 def markdown(source, mentions=False):
     return literal('<div class="markdown-block">%s</div>' %
                    MarkupRenderer.markdown(source, flavored=True,
                                            mentions=mentions))
 
 def renderer_from_filename(filename, exclude=None):
     return MarkupRenderer.renderer_from_filename(filename, exclude=exclude)
 
 
 def render(source, renderer='rst', mentions=False):
     if renderer == 'rst':
         return rst(source, mentions=mentions)
     if renderer == 'markdown':
         return markdown(source, mentions=mentions)
 
 
 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():
     return _(
         'Example filter terms:\n' +
         '     repository:vcs\n' +
         '     username: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 not_mapped_error(repo_name):
     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(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
     form = insecure_form(url, method, multipart, **attrs)
     token = csrf_input()
     return literal("%s\n%s" % (form, token))
 
 def csrf_input():
     return literal(
         '<input type="hidden" id="{}" name="{}" value="{}">'.format(
         csrf_token_key, csrf_token_key, get_csrf_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_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.route_path(*args, **kwds)
 
 
 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)

rhodecode/lib/middleware/simplesvn.py

@@ -1,139 +1,159 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  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 logging
 from urlparse import urljoin
 
 import requests
+from webob.exc import HTTPNotAcceptable
 
-import rhodecode
 from rhodecode.lib.middleware import simplevcs
 from rhodecode.lib.utils import is_valid_repo
+from rhodecode.lib.utils2 import str2bool
 
 log = logging.getLogger(__name__)
 
 
 class SimpleSvnApp(object):
     IGNORED_HEADERS = [
         'connection', 'keep-alive', 'content-encoding',
         'transfer-encoding', 'content-length']
 
     def __init__(self, config):
         self.config = config
 
     def __call__(self, environ, start_response):
         request_headers = self._get_request_headers(environ)
 
         data = environ['wsgi.input']
         # johbo: Avoid that we end up with sending the request in chunked
         # transfer encoding (mainly on Gunicorn). If we know the content
         # length, then we should transfer the payload in one request.
         if environ['REQUEST_METHOD'] == 'MKCOL' or 'CONTENT_LENGTH' in environ:
             data = data.read()
 
         response = requests.request(
             environ['REQUEST_METHOD'], self._get_url(environ['PATH_INFO']),
             data=data, headers=request_headers)
 
         response_headers = self._get_response_headers(response.headers)
         start_response(
             '{} {}'.format(response.status_code, response.reason),
             response_headers)
         return response.iter_content(chunk_size=1024)
 
     def _get_url(self, path):
         return urljoin(
             self.config.get('subversion_http_server_url', ''), path)
 
     def _get_request_headers(self, environ):
         headers = {}
 
         for key in environ:
             if not key.startswith('HTTP_'):
                 continue
             new_key = key.split('_')
             new_key = [k.capitalize() for k in new_key[1:]]
             new_key = '-'.join(new_key)
             headers[new_key] = environ[key]
 
         if 'CONTENT_TYPE' in environ:
             headers['Content-Type'] = environ['CONTENT_TYPE']
 
         if 'CONTENT_LENGTH' in environ:
             headers['Content-Length'] = environ['CONTENT_LENGTH']
 
         return headers
 
     def _get_response_headers(self, headers):
         headers = [
             (h, headers[h])
             for h in headers
             if h.lower() not in self.IGNORED_HEADERS
         ]
 
         # Add custom response header to indicate that this is a VCS response
         # and which backend is used.
         headers.append(('X-RhodeCode-Backend', 'svn'))
 
         return headers
 
 
+class DisabledSimpleSvnApp(object):
+    def __init__(self, config):
+        self.config = config
+
+    def __call__(self, environ, start_response):
+        reason = 'Cannot handle SVN call because: SVN HTTP Proxy is not enabled'
+        log.warning(reason)
+        return HTTPNotAcceptable(reason)(environ, start_response)
+
+
 class SimpleSvn(simplevcs.SimpleVCS):
 
     SCM = 'svn'
     READ_ONLY_COMMANDS = ('OPTIONS', 'PROPFIND', 'GET', 'REPORT')
+    DEFAULT_HTTP_SERVER = 'http://localhost:8090'
 
     def _get_repository_name(self, environ):
         """
         Gets repository name out of PATH_INFO header
 
         :param environ: environ where PATH_INFO is stored
         """
         path = environ['PATH_INFO'].split('!')
         repo_name = path[0].strip('/')
 
         # SVN includes the whole path in it's requests, including
         # subdirectories inside the repo. Therefore we have to search for
         # the repo root directory.
         if not is_valid_repo(repo_name, self.basepath, self.SCM):
             current_path = ''
             for component in repo_name.split('/'):
                 current_path += component
                 if is_valid_repo(current_path, self.basepath, self.SCM):
                     return current_path
                 current_path += '/'
 
         return repo_name
 
     def _get_action(self, environ):
         return (
             'pull'
             if environ['REQUEST_METHOD'] in self.READ_ONLY_COMMANDS
             else 'push')
 
     def _create_wsgi_app(self, repo_path, repo_name, config):
+        if self._is_svn_enabled():
         return SimpleSvnApp(config)
+        # we don't have http proxy enabled return dummy request handler
+        return DisabledSimpleSvnApp(config)
+
+    def _is_svn_enabled(self):
+        conf = self.repo_vcs_config
+        return str2bool(conf.get('vcs_svn_proxy', 'http_requests_enabled'))
 
     def _create_config(self, extras, repo_name):
-        server_url = rhodecode.CONFIG.get(
-            'rhodecode_subversion_http_server_url', '')
-        extras['subversion_http_server_url'] = (
-            server_url or 'http://localhost/')
+        conf = self.repo_vcs_config
+        server_url = conf.get('vcs_svn_proxy', 'http_server_url')
+        server_url = server_url or self.DEFAULT_HTTP_SERVER
+
+        extras['subversion_http_server_url'] = server_url
         return extras

rhodecode/lib/middleware/simplevcs.py

@@ -1,448 +1,452 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2014-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 SimpleVCS middleware for handling protocol request (push/clone etc.)
 It's implemented with basic auth function
 """
 
 import os
 import logging
 import importlib
 from functools import wraps
 
 from paste.httpheaders import REMOTE_USER, AUTH_TYPE
 from webob.exc import (
     HTTPNotFound, HTTPForbidden, HTTPNotAcceptable, HTTPInternalServerError)
 
 import rhodecode
 from rhodecode.authentication.base import authenticate, VCS_TYPE
 from rhodecode.lib.auth import AuthUser, HasPermissionAnyMiddleware
 from rhodecode.lib.base import BasicAuth, get_ip_addr, vcs_operation_context
 from rhodecode.lib.exceptions import (
     HTTPLockedRC, HTTPRequirementError, UserCreationError,
     NotAllowedToCreateUserError)
 from rhodecode.lib.hooks_daemon import prepare_callback_daemon
 from rhodecode.lib.middleware import appenlight
 from rhodecode.lib.middleware.utils import scm_app
 from rhodecode.lib.utils import (
     is_valid_repo, get_rhodecode_realm, get_rhodecode_base_path)
 from rhodecode.lib.utils2 import safe_str, fix_PATH, str2bool
 from rhodecode.lib.vcs.conf import settings as vcs_settings
+from rhodecode.lib.vcs.backends import base
 from rhodecode.model import meta
 from rhodecode.model.db import User, Repository
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.settings import SettingsModel
 
+
 log = logging.getLogger(__name__)
 
 
 def initialize_generator(factory):
     """
     Initializes the returned generator by draining its first element.
 
     This can be used to give a generator an initializer, which is the code
     up to the first yield statement. This decorator enforces that the first
     produced element has the value ``"__init__"`` to make its special
     purpose very explicit in the using code.
     """
 
     @wraps(factory)
     def wrapper(*args, **kwargs):
         gen = factory(*args, **kwargs)
         try:
             init = gen.next()
         except StopIteration:
             raise ValueError('Generator must yield at least one element.')
         if init != "__init__":
             raise ValueError('First yielded element must be "__init__".')
         return gen
     return wrapper
 
 
 class SimpleVCS(object):
     """Common functionality for SCM HTTP handlers."""
 
     SCM = 'unknown'
 
     def __init__(self, application, config, registry):
         self.registry = registry
         self.application = application
         self.config = config
+        # re-populated by specialized middlewares
+        self.repo_vcs_config = base.Config()
         # base path of repo locations
         self.basepath = get_rhodecode_base_path()
         # authenticate this VCS request using authfunc
         auth_ret_code_detection = \
             str2bool(self.config.get('auth_ret_code_detection', False))
         self.authenticate = BasicAuth(
             '', authenticate, registry, config.get('auth_ret_code'),
             auth_ret_code_detection)
         self.ip_addr = '0.0.0.0'
 
     @property
     def scm_app(self):
         custom_implementation = self.config.get('vcs.scm_app_implementation')
         if custom_implementation and custom_implementation != 'pyro4':
             log.info(
                 "Using custom implementation of scm_app: %s",
                 custom_implementation)
             scm_app_impl = importlib.import_module(custom_implementation)
         else:
             scm_app_impl = scm_app
         return scm_app_impl
 
     def _get_by_id(self, repo_name):
         """
         Gets a special pattern _<ID> from clone url and tries to replace it
         with a repository_name for support of _<ID> non changable urls
 
         :param repo_name:
         """
 
         data = repo_name.split('/')
         if len(data) >= 2:
             from rhodecode.model.repo import RepoModel
             by_id_match = RepoModel().get_repo_by_id(repo_name)
             if by_id_match:
                 data[1] = by_id_match.repo_name
 
         return safe_str('/'.join(data))
 
     def _invalidate_cache(self, repo_name):
         """
         Set's cache for this repository for invalidation on next access
 
         :param repo_name: full repo name, also a cache key
         """
         ScmModel().mark_for_invalidation(repo_name)
 
     def is_valid_and_existing_repo(self, repo_name, base_path, scm_type):
         db_repo = Repository.get_by_repo_name(repo_name)
         if not db_repo:
             log.debug('Repository `%s` not found inside the database.',
                       repo_name)
             return False
 
         if db_repo.repo_type != scm_type:
             log.warning(
                 'Repository `%s` have incorrect scm_type, expected %s got %s',
                 repo_name, db_repo.repo_type, scm_type)
             return False
 
         return is_valid_repo(repo_name, base_path, expect_scm=scm_type)
 
     def valid_and_active_user(self, user):
         """
         Checks if that user is not empty, and if it's actually object it checks
         if he's active.
 
         :param user: user object or None
         :return: boolean
         """
         if user is None:
             return False
 
         elif user.active:
             return True
 
         return False
 
     def _check_permission(self, action, user, repo_name, ip_addr=None):
         """
         Checks permissions using action (push/pull) user and repository
         name
 
         :param action: push or pull action
         :param user: user instance
         :param repo_name: repository name
         """
         # check IP
         inherit = user.inherit_default_permissions
         ip_allowed = AuthUser.check_ip_allowed(user.user_id, ip_addr,
                                                inherit_from_default=inherit)
         if ip_allowed:
             log.info('Access for IP:%s allowed', ip_addr)
         else:
             return False
 
         if action == 'push':
             if not HasPermissionAnyMiddleware('repository.write',
                                               'repository.admin')(user,
                                                                   repo_name):
                 return False
 
         else:
             # any other action need at least read permission
             if not HasPermissionAnyMiddleware('repository.read',
                                               'repository.write',
                                               'repository.admin')(user,
                                                                   repo_name):
                 return False
 
         return True
 
     def _check_ssl(self, environ, start_response):
         """
         Checks the SSL check flag and returns False if SSL is not present
         and required True otherwise
         """
         org_proto = environ['wsgi._org_proto']
         # check if we have SSL required  ! if not it's a bad request !
         require_ssl = str2bool(
             SettingsModel().get_ui_by_key('push_ssl').ui_value)
         if require_ssl and org_proto == 'http':
             log.debug('proto is %s and SSL is required BAD REQUEST !',
                       org_proto)
             return False
         return True
 
     def __call__(self, environ, start_response):
         try:
             return self._handle_request(environ, start_response)
         except Exception:
             log.exception("Exception while handling request")
             appenlight.track_exception(environ)
             return HTTPInternalServerError()(environ, start_response)
         finally:
             meta.Session.remove()
 
     def _handle_request(self, environ, start_response):
 
         if not self._check_ssl(environ, start_response):
             reason = ('SSL required, while RhodeCode was unable '
                       'to detect this as SSL request')
             log.debug('User not allowed to proceed, %s', reason)
             return HTTPNotAcceptable(reason)(environ, start_response)
 
         ip_addr = get_ip_addr(environ)
         username = None
 
         # skip passing error to error controller
         environ['pylons.status_code_redirect'] = True
 
         # ======================================================================
         # EXTRACT REPOSITORY NAME FROM ENV
         # ======================================================================
         environ['PATH_INFO'] = self._get_by_id(environ['PATH_INFO'])
         repo_name = self._get_repository_name(environ)
         environ['REPO_NAME'] = repo_name
         log.debug('Extracted repo name is %s', repo_name)
 
         # check for type, presence in database and on filesystem
         if not self.is_valid_and_existing_repo(
                 repo_name, self.basepath, self.SCM):
             return HTTPNotFound()(environ, start_response)
 
         # ======================================================================
         # GET ACTION PULL or PUSH
         # ======================================================================
         action = self._get_action(environ)
 
         # ======================================================================
         # CHECK ANONYMOUS PERMISSION
         # ======================================================================
         if action in ['pull', 'push']:
             anonymous_user = User.get_default_user()
             username = anonymous_user.username
             if anonymous_user.active:
                 # ONLY check permissions if the user is activated
                 anonymous_perm = self._check_permission(
                     action, anonymous_user, repo_name, ip_addr)
             else:
                 anonymous_perm = False
 
             if not anonymous_user.active or not anonymous_perm:
                 if not anonymous_user.active:
                     log.debug('Anonymous access is disabled, running '
                               'authentication')
 
                 if not anonymous_perm:
                     log.debug('Not enough credentials to access this '
                               'repository as anonymous user')
 
                 username = None
                 # ==============================================================
                 # DEFAULT PERM FAILED OR ANONYMOUS ACCESS IS DISABLED SO WE
                 # NEED TO AUTHENTICATE AND ASK FOR AUTH USER PERMISSIONS
                 # ==============================================================
 
                 # try to auth based on environ, container auth methods
                 log.debug('Running PRE-AUTH for container based authentication')
                 pre_auth = authenticate(
                     '', '', environ, VCS_TYPE, registry=self.registry)
                 if pre_auth and pre_auth.get('username'):
                     username = pre_auth['username']
                 log.debug('PRE-AUTH got %s as username', username)
 
                 # If not authenticated by the container, running basic auth
                 if not username:
                     self.authenticate.realm = get_rhodecode_realm()
 
                     try:
                         result = self.authenticate(environ)
                     except (UserCreationError, NotAllowedToCreateUserError) as e:
                         log.error(e)
                         reason = safe_str(e)
                         return HTTPNotAcceptable(reason)(environ, start_response)
 
                     if isinstance(result, str):
                         AUTH_TYPE.update(environ, 'basic')
                         REMOTE_USER.update(environ, result)
                         username = result
                     else:
                         return result.wsgi_application(environ, start_response)
 
                 # ==============================================================
                 # CHECK PERMISSIONS FOR THIS REQUEST USING GIVEN USERNAME
                 # ==============================================================
                 user = User.get_by_username(username)
                 if not self.valid_and_active_user(user):
                     return HTTPForbidden()(environ, start_response)
                 username = user.username
                 user.update_lastactivity()
                 meta.Session().commit()
 
                 # check user attributes for password change flag
                 user_obj = user
                 if user_obj and user_obj.username != User.DEFAULT_USER and \
                         user_obj.user_data.get('force_password_change'):
                     reason = 'password change required'
                     log.debug('User not allowed to authenticate, %s', reason)
                     return HTTPNotAcceptable(reason)(environ, start_response)
 
                 # check permissions for this repository
                 perm = self._check_permission(action, user, repo_name, ip_addr)
                 if not perm:
                     return HTTPForbidden()(environ, start_response)
 
         # extras are injected into UI object and later available
         # in hooks executed by rhodecode
         check_locking = _should_check_locking(environ.get('QUERY_STRING'))
         extras = vcs_operation_context(
             environ, repo_name=repo_name, username=username,
             action=action, scm=self.SCM,
             check_locking=check_locking)
 
         # ======================================================================
         # REQUEST HANDLING
         # ======================================================================
         str_repo_name = safe_str(repo_name)
         repo_path = os.path.join(safe_str(self.basepath), str_repo_name)
         log.debug('Repository path is %s', repo_path)
 
         fix_PATH()
 
         log.info(
             '%s action on %s repo "%s" by "%s" from %s',
             action, self.SCM, str_repo_name, safe_str(username), ip_addr)
 
         return self._generate_vcs_response(
             environ, start_response, repo_path, repo_name, extras, action)
 
     @initialize_generator
     def _generate_vcs_response(
             self, environ, start_response, repo_path, repo_name, extras,
             action):
         """
         Returns a generator for the response content.
 
         This method is implemented as a generator, so that it can trigger
         the cache validation after all content sent back to the client. It
         also handles the locking exceptions which will be triggered when
         the first chunk is produced by the underlying WSGI application.
         """
         callback_daemon, extras = self._prepare_callback_daemon(extras)
         config = self._create_config(extras, repo_name)
         log.debug('HOOKS extras is %s', extras)
         app = self._create_wsgi_app(repo_path, repo_name, config)
 
         try:
             with callback_daemon:
                 try:
                     response = app(environ, start_response)
                 finally:
                     # This statement works together with the decorator
                     # "initialize_generator" above. The decorator ensures that
                     # we hit the first yield statement before the generator is
                     # returned back to the WSGI server. This is needed to
                     # ensure that the call to "app" above triggers the
                     # needed callback to "start_response" before the
                     # generator is actually used.
                     yield "__init__"
 
                 for chunk in response:
                     yield chunk
         except Exception as exc:
             # TODO: johbo: Improve "translating" back the exception.
             if getattr(exc, '_vcs_kind', None) == 'repo_locked':
                 exc = HTTPLockedRC(*exc.args)
                 _code = rhodecode.CONFIG.get('lock_ret_code')
                 log.debug('Repository LOCKED ret code %s!', (_code,))
             elif getattr(exc, '_vcs_kind', None) == 'requirement':
                 log.debug(
                     'Repository requires features unknown to this Mercurial')
                 exc = HTTPRequirementError(*exc.args)
             else:
                 raise
 
             for chunk in exc(environ, start_response):
                 yield chunk
         finally:
             # invalidate cache on push
             try:
                 if action == 'push':
                     self._invalidate_cache(repo_name)
             finally:
                 meta.Session.remove()
 
     def _get_repository_name(self, environ):
         """Get repository name out of the environmnent
 
         :param environ: WSGI environment
         """
         raise NotImplementedError()
 
     def _get_action(self, environ):
         """Map request commands into a pull or push command.
 
         :param environ: WSGI environment
         """
         raise NotImplementedError()
 
     def _create_wsgi_app(self, repo_path, repo_name, config):
         """Return the WSGI app that will finally handle the request."""
         raise NotImplementedError()
 
     def _create_config(self, extras, repo_name):
         """Create a Pyro safe config representation."""
         raise NotImplementedError()
 
     def _prepare_callback_daemon(self, extras):
         return prepare_callback_daemon(
             extras, protocol=vcs_settings.HOOKS_PROTOCOL,
             use_direct_calls=vcs_settings.HOOKS_DIRECT_CALLS)
 
 
 def _should_check_locking(query_string):
     # this is kind of hacky, but due to how mercurial handles client-server
     # server see all operation on commit; bookmarks, phases and
     # obsolescence marker in different transaction, we don't want to check
     # locking on those
     return query_string not in ['cmd=listkeys']

rhodecode/lib/middleware/vcs.py

@@ -1,161 +1,178 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  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 gzip
 import shutil
 import logging
 import tempfile
 import urlparse
 
 import rhodecode
 from rhodecode.lib.middleware.appenlight import wrap_in_appenlight_if_enabled
 from rhodecode.lib.middleware.simplegit import SimpleGit, GIT_PROTO_PAT
 from rhodecode.lib.middleware.simplehg import SimpleHg
 from rhodecode.lib.middleware.simplesvn import SimpleSvn
-
+from rhodecode.lib.vcs.backends import base
+from rhodecode.model.settings import VcsSettingsModel
 
 log = logging.getLogger(__name__)
 
 
 def is_git(environ):
     """
     Returns True if requests should be handled by GIT wsgi middleware
     """
     is_git_path = GIT_PROTO_PAT.match(environ['PATH_INFO'])
     log.debug(
         'request path: `%s` detected as GIT PROTOCOL %s', environ['PATH_INFO'],
         is_git_path is not None)
 
     return is_git_path
 
 
 def is_hg(environ):
     """
     Returns True if requests target is mercurial server - header
     ``HTTP_ACCEPT`` of such request would start with ``application/mercurial``.
     """
     is_hg_path = False
 
     http_accept = environ.get('HTTP_ACCEPT')
 
     if http_accept and http_accept.startswith('application/mercurial'):
         query = urlparse.parse_qs(environ['QUERY_STRING'])
         if 'cmd' in query:
             is_hg_path = True
 
     log.debug(
         'request path: `%s` detected as HG PROTOCOL %s', environ['PATH_INFO'],
         is_hg_path)
 
     return is_hg_path
 
 
 def is_svn(environ):
     """
     Returns True if requests target is Subversion server
     """
     http_dav = environ.get('HTTP_DAV', '')
     magic_path_segment = rhodecode.CONFIG.get(
         'rhodecode_subversion_magic_path', '/!svn')
     is_svn_path = (
         'subversion' in http_dav or
         magic_path_segment in environ['PATH_INFO'])
     log.debug(
         'request path: `%s` detected as SVN PROTOCOL %s', environ['PATH_INFO'],
         is_svn_path)
 
     return is_svn_path
 
 
 class GunzipMiddleware(object):
     """
     WSGI middleware that unzips gzip-encoded requests before
     passing on to the underlying application.
     """
 
     def __init__(self, application):
         self.app = application
 
     def __call__(self, environ, start_response):
         accepts_encoding_header = environ.get('HTTP_CONTENT_ENCODING', b'')
 
         if b'gzip' in accepts_encoding_header:
             log.debug('gzip detected, now running gunzip wrapper')
             wsgi_input = environ['wsgi.input']
 
             if not hasattr(environ['wsgi.input'], 'seek'):
                 # The gzip implementation in the standard library of Python 2.x
                 # requires the '.seek()' and '.tell()' methods to be available
                 # on the input stream.  Read the data into a temporary file to
                 # work around this limitation.
 
                 wsgi_input = tempfile.SpooledTemporaryFile(64 * 1024 * 1024)
                 shutil.copyfileobj(environ['wsgi.input'], wsgi_input)
                 wsgi_input.seek(0)
 
             environ['wsgi.input'] = gzip.GzipFile(fileobj=wsgi_input, mode='r')
             # since we "Ungzipped" the content we say now it's no longer gzip
             # content encoding
             del environ['HTTP_CONTENT_ENCODING']
 
             # content length has changes ? or i'm not sure
             if 'CONTENT_LENGTH' in environ:
                 del environ['CONTENT_LENGTH']
         else:
             log.debug('content not gzipped, gzipMiddleware passing '
                       'request further')
         return self.app(environ, start_response)
 
 
 class VCSMiddleware(object):
 
     def __init__(self, app, config, appenlight_client, registry):
         self.application = app
         self.config = config
         self.appenlight_client = appenlight_client
         self.registry = registry
+        self.repo_vcs_config = base.Config()
+        self.use_gzip = True
+
+    def vcs_config(self, repo_name=None):
+        """
+        returns serialized VcsSettings
+        """
+        return VcsSettingsModel(repo=repo_name).get_ui_settings_as_config_obj()
+
+    def wrap_in_gzip_if_enabled(self, app):
+        if self.use_gzip:
+            app = GunzipMiddleware(app)
+        return app
 
     def _get_handler_app(self, environ):
         app = None
+
         if is_hg(environ):
             app = SimpleHg(self.application, self.config, self.registry)
 
         if is_git(environ):
             app = SimpleGit(self.application, self.config, self.registry)
 
-        proxy_svn = rhodecode.CONFIG.get(
-            'rhodecode_proxy_subversion_http_requests', False)
-        if proxy_svn and is_svn(environ):
+        if is_svn(environ):
             app = SimpleSvn(self.application, self.config, self.registry)
 
         if app:
-            app = GunzipMiddleware(app)
+            repo_name = app._get_repository_name(environ)
+            self.repo_vcs_config = self.vcs_config(repo_name)
+            app.repo_vcs_config = self.repo_vcs_config
+
+            app = self.wrap_in_gzip_if_enabled(app)
             app, _ = wrap_in_appenlight_if_enabled(
                 app, self.config, self.appenlight_client)
 
         return app
 
     def __call__(self, environ, start_response):
         # check if we handle one of interesting protocols ?
         vcs_handler = self._get_handler_app(environ)
         if vcs_handler:
             return vcs_handler(environ, start_response)
 
         return self.application(environ, start_response)

rhodecode/model/forms.py

@@ -1,547 +1,547 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  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/
 
 """
 this is forms validation classes
 http://formencode.org/module-formencode.validators.html
 for list off all availible validators
 
 we can create our own validators
 
 The table below outlines the options which can be used in a schema in addition to the validators themselves
 pre_validators          []     These validators will be applied before the schema
 chained_validators      []     These validators will be applied after the schema
 allow_extra_fields      False     If True, then it is not an error when keys that aren't associated with a validator are present
 filter_extra_fields     False     If True, then keys that aren't associated with a validator are removed
 if_key_missing          NoDefault If this is given, then any keys that aren't available but are expected will be replaced with this value (and then validated). This does not override a present .if_missing attribute on validators. NoDefault is a special FormEncode class to mean that no default values has been specified and therefore missing keys shouldn't take a default value.
 ignore_key_missing      False     If True, then missing keys will be missing in the result, if the validator doesn't have .if_missing on it already
 
 
 <name> = formencode.validators.<name of validator>
 <name> must equal form name
 list=[1,2,3,4,5]
 for SELECT use formencode.All(OneOf(list), Int())
 
 """
 
 import deform
 import logging
 import formencode
 
 from pkg_resources import resource_filename
 from formencode import All, Pipe
 
 from pylons.i18n.translation import _
 
 from rhodecode import BACKENDS
 from rhodecode.lib import helpers
 from rhodecode.model import validators as v
 
 log = logging.getLogger(__name__)
 
 
 deform_templates = resource_filename('deform', 'templates')
 rhodecode_templates = resource_filename('rhodecode', 'templates/forms')
 search_path = (rhodecode_templates, deform_templates)
 
 
 class RhodecodeFormZPTRendererFactory(deform.ZPTRendererFactory):
     """ Subclass of ZPTRendererFactory to add rhodecode context variables """
     def __call__(self, template_name, **kw):
         kw['h'] = helpers
         return self.load(template_name)(**kw)
 
 
 form_renderer = RhodecodeFormZPTRendererFactory(search_path)
 deform.Form.set_default_renderer(form_renderer)
 
 
 def LoginForm():
     class _LoginForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = v.UnicodeString(
             strip=True,
             min=1,
             not_empty=True,
             messages={
                 'empty': _(u'Please enter a login'),
                 'tooShort': _(u'Enter a value %(min)i characters long or more')
             }
         )
 
         password = v.UnicodeString(
             strip=False,
             min=3,
             not_empty=True,
             messages={
                 'empty': _(u'Please enter a password'),
                 'tooShort': _(u'Enter %(min)i characters or more')}
         )
 
         remember = v.StringBoolean(if_missing=False)
 
         chained_validators = [v.ValidAuth()]
     return _LoginForm
 
 
 def UserForm(edit=False, available_languages=[], old_data={}):
     class _UserForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                        v.ValidUsername(edit, old_data))
         if edit:
             new_password = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=False)
             )
             password_confirmation = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=False),
             )
             admin = v.StringBoolean(if_missing=False)
         else:
             password = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=True)
             )
             password_confirmation = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=False)
             )
 
         password_change = v.StringBoolean(if_missing=False)
         create_repo_group = v.StringBoolean(if_missing=False)
 
         active = v.StringBoolean(if_missing=False)
         firstname = v.UnicodeString(strip=True, min=1, not_empty=False)
         lastname = v.UnicodeString(strip=True, min=1, not_empty=False)
         email = All(v.Email(not_empty=True), v.UniqSystemEmail(old_data))
         extern_name = v.UnicodeString(strip=True)
         extern_type = v.UnicodeString(strip=True)
         language = v.OneOf(available_languages, hideList=False,
                            testValueList=True, if_missing=None)
         chained_validators = [v.ValidPasswordsMatch()]
     return _UserForm
 
 
 def UserGroupForm(edit=False, old_data=None, available_members=None,
                   allow_disabled=False):
     old_data = old_data or {}
     available_members = available_members or []
 
     class _UserGroupForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         users_group_name = All(
             v.UnicodeString(strip=True, min=1, not_empty=True),
             v.ValidUserGroup(edit, old_data)
         )
         user_group_description = v.UnicodeString(strip=True, min=1,
                                                  not_empty=False)
 
         users_group_active = v.StringBoolean(if_missing=False)
 
         if edit:
             users_group_members = v.OneOf(
                 available_members, hideList=False, testValueList=True,
                 if_missing=None, not_empty=False
             )
             # this is user group owner
             user = All(
                 v.UnicodeString(not_empty=True),
                 v.ValidRepoUser(allow_disabled))
     return _UserGroupForm
 
 
 def RepoGroupForm(edit=False, old_data=None, available_groups=None,
                    can_create_in_root=False, allow_disabled=False):
     old_data = old_data or {}
     available_groups = available_groups or []
 
     class _RepoGroupForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
 
         group_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                          v.SlugifyName(),)
         group_description = v.UnicodeString(strip=True, min=1,
                                             not_empty=False)
         group_copy_permissions = v.StringBoolean(if_missing=False)
 
         group_parent_id = v.OneOf(available_groups, hideList=False,
                                   testValueList=True, not_empty=True)
         enable_locking = v.StringBoolean(if_missing=False)
         chained_validators = [
             v.ValidRepoGroup(edit, old_data, can_create_in_root)]
 
         if edit:
             # this is repo group owner
             user = All(
                 v.UnicodeString(not_empty=True),
                 v.ValidRepoUser(allow_disabled))
 
     return _RepoGroupForm
 
 
 def RegisterForm(edit=False, old_data={}):
     class _RegisterForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = All(
             v.ValidUsername(edit, old_data),
             v.UnicodeString(strip=True, min=1, not_empty=True)
         )
         password = All(
             v.ValidPassword(),
             v.UnicodeString(strip=False, min=6, not_empty=True)
         )
         password_confirmation = All(
             v.ValidPassword(),
             v.UnicodeString(strip=False, min=6, not_empty=True)
         )
         active = v.StringBoolean(if_missing=False)
         firstname = v.UnicodeString(strip=True, min=1, not_empty=False)
         lastname = v.UnicodeString(strip=True, min=1, not_empty=False)
         email = All(v.Email(not_empty=True), v.UniqSystemEmail(old_data))
 
         chained_validators = [v.ValidPasswordsMatch()]
 
     return _RegisterForm
 
 
 def PasswordResetForm():
     class _PasswordResetForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         email = All(v.ValidSystemEmail(), v.Email(not_empty=True))
     return _PasswordResetForm
 
 
 def RepoForm(edit=False, old_data=None, repo_groups=None, landing_revs=None,
              allow_disabled=False):
     old_data = old_data or {}
     repo_groups = repo_groups or []
     landing_revs = landing_revs or []
     supported_backends = BACKENDS.keys()
 
     class _RepoForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         repo_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                         v.SlugifyName())
         repo_group = All(v.CanWriteGroup(old_data),
                          v.OneOf(repo_groups, hideList=True))
         repo_type = v.OneOf(supported_backends, required=False,
                             if_missing=old_data.get('repo_type'))
         repo_description = v.UnicodeString(strip=True, min=1, not_empty=False)
         repo_private = v.StringBoolean(if_missing=False)
         repo_landing_rev = v.OneOf(landing_revs, hideList=True)
         repo_copy_permissions = v.StringBoolean(if_missing=False)
         clone_uri = All(v.UnicodeString(strip=True, min=1, not_empty=False))
 
         repo_enable_statistics = v.StringBoolean(if_missing=False)
         repo_enable_downloads = v.StringBoolean(if_missing=False)
         repo_enable_locking = v.StringBoolean(if_missing=False)
 
         if edit:
             # this is repo owner
             user = All(
                 v.UnicodeString(not_empty=True),
                 v.ValidRepoUser(allow_disabled))
             clone_uri_change = v.UnicodeString(
                 not_empty=False, if_missing=v.Missing)
 
         chained_validators = [v.ValidCloneUri(),
                               v.ValidRepoName(edit, old_data)]
     return _RepoForm
 
 
 def RepoPermsForm():
     class _RepoPermsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         chained_validators = [v.ValidPerms(type_='repo')]
     return _RepoPermsForm
 
 
 def RepoGroupPermsForm(valid_recursive_choices):
     class _RepoGroupPermsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         recursive = v.OneOf(valid_recursive_choices)
         chained_validators = [v.ValidPerms(type_='repo_group')]
     return _RepoGroupPermsForm
 
 
 def UserGroupPermsForm():
     class _UserPermsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         chained_validators = [v.ValidPerms(type_='user_group')]
     return _UserPermsForm
 
 
 def RepoFieldForm():
     class _RepoFieldForm(formencode.Schema):
         filter_extra_fields = True
         allow_extra_fields = True
 
         new_field_key = All(v.FieldKey(),
                             v.UnicodeString(strip=True, min=3, not_empty=True))
         new_field_value = v.UnicodeString(not_empty=False, if_missing=u'')
         new_field_type = v.OneOf(['str', 'unicode', 'list', 'tuple'],
                                  if_missing='str')
         new_field_label = v.UnicodeString(not_empty=False)
         new_field_desc = v.UnicodeString(not_empty=False)
 
     return _RepoFieldForm
 
 
 def RepoForkForm(edit=False, old_data={}, supported_backends=BACKENDS.keys(),
                  repo_groups=[], landing_revs=[]):
     class _RepoForkForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         repo_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                         v.SlugifyName())
         repo_group = All(v.CanWriteGroup(),
                          v.OneOf(repo_groups, hideList=True))
         repo_type = All(v.ValidForkType(old_data), v.OneOf(supported_backends))
         description = v.UnicodeString(strip=True, min=1, not_empty=True)
         private = v.StringBoolean(if_missing=False)
         copy_permissions = v.StringBoolean(if_missing=False)
         fork_parent_id = v.UnicodeString()
         chained_validators = [v.ValidForkName(edit, old_data)]
         landing_rev = v.OneOf(landing_revs, hideList=True)
 
     return _RepoForkForm
 
 
 def ApplicationSettingsForm():
     class _ApplicationSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         rhodecode_title = v.UnicodeString(strip=True, max=40, not_empty=False)
         rhodecode_realm = v.UnicodeString(strip=True, min=1, not_empty=True)
         rhodecode_pre_code = v.UnicodeString(strip=True, min=1, not_empty=False)
         rhodecode_post_code = v.UnicodeString(strip=True, min=1, not_empty=False)
         rhodecode_captcha_public_key = v.UnicodeString(strip=True, min=1, not_empty=False)
         rhodecode_captcha_private_key = v.UnicodeString(strip=True, min=1, not_empty=False)
 
     return _ApplicationSettingsForm
 
 
 def ApplicationVisualisationForm():
     class _ApplicationVisualisationForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         rhodecode_show_public_icon = v.StringBoolean(if_missing=False)
         rhodecode_show_private_icon = v.StringBoolean(if_missing=False)
         rhodecode_stylify_metatags = v.StringBoolean(if_missing=False)
 
         rhodecode_repository_fields = v.StringBoolean(if_missing=False)
         rhodecode_lightweight_journal = v.StringBoolean(if_missing=False)
         rhodecode_dashboard_items = v.Int(min=5, not_empty=True)
         rhodecode_admin_grid_items = v.Int(min=5, not_empty=True)
         rhodecode_show_version = v.StringBoolean(if_missing=False)
         rhodecode_use_gravatar = v.StringBoolean(if_missing=False)
         rhodecode_markup_renderer = v.OneOf(['markdown', 'rst'])
         rhodecode_gravatar_url = v.UnicodeString(min=3)
         rhodecode_clone_uri_tmpl = v.UnicodeString(min=3)
         rhodecode_support_url = v.UnicodeString()
         rhodecode_show_revision_number = v.StringBoolean(if_missing=False)
         rhodecode_show_sha_length = v.Int(min=4, not_empty=True)
 
     return _ApplicationVisualisationForm
 
 
 class _BaseVcsSettingsForm(formencode.Schema):
     allow_extra_fields = True
     filter_extra_fields = False
     hooks_changegroup_repo_size = v.StringBoolean(if_missing=False)
     hooks_changegroup_push_logger = v.StringBoolean(if_missing=False)
     hooks_outgoing_pull_logger = v.StringBoolean(if_missing=False)
 
     extensions_largefiles = v.StringBoolean(if_missing=False)
     phases_publish = v.StringBoolean(if_missing=False)
 
     rhodecode_pr_merge_enabled = v.StringBoolean(if_missing=False)
     rhodecode_use_outdated_comments = v.StringBoolean(if_missing=False)
     rhodecode_hg_use_rebase_for_merging = v.StringBoolean(if_missing=False)
 
-    rhodecode_proxy_subversion_http_requests = v.StringBoolean(if_missing=False)
-    rhodecode_subversion_http_server_url = v.UnicodeString(
-        strip=True, if_missing=None)
+    vcs_svn_proxy_http_requests_enabled = v.StringBoolean(if_missing=False)
+    vcs_svn_proxy_http_server_url = v.UnicodeString(strip=True, if_missing=None)
+
 
 def ApplicationUiSettingsForm():
     class _ApplicationUiSettingsForm(_BaseVcsSettingsForm):
         web_push_ssl = v.StringBoolean(if_missing=False)
         paths_root_path = All(
             v.ValidPath(),
             v.UnicodeString(strip=True, min=1, not_empty=True)
         )
         extensions_hgsubversion = v.StringBoolean(if_missing=False)
         extensions_hggit = v.StringBoolean(if_missing=False)
         new_svn_branch = v.ValidSvnPattern(section='vcs_svn_branch')
         new_svn_tag = v.ValidSvnPattern(section='vcs_svn_tag')
 
     return _ApplicationUiSettingsForm
 
 
 def RepoVcsSettingsForm(repo_name):
     class _RepoVcsSettingsForm(_BaseVcsSettingsForm):
         inherit_global_settings = v.StringBoolean(if_missing=False)
         new_svn_branch = v.ValidSvnPattern(
             section='vcs_svn_branch', repo_name=repo_name)
         new_svn_tag = v.ValidSvnPattern(
             section='vcs_svn_tag', repo_name=repo_name)
 
     return _RepoVcsSettingsForm
 
 
 def LabsSettingsForm():
     class _LabSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
 
     return _LabSettingsForm
 
 
 def ApplicationPermissionsForm(register_choices, extern_activate_choices):
     class _DefaultPermissionsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         anonymous = v.StringBoolean(if_missing=False)
         default_register = v.OneOf(register_choices)
         default_register_message = v.UnicodeString()
         default_extern_activate = v.OneOf(extern_activate_choices)
 
     return _DefaultPermissionsForm
 
 
 def ObjectPermissionsForm(repo_perms_choices, group_perms_choices,
                           user_group_perms_choices):
     class _ObjectPermissionsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         overwrite_default_repo = v.StringBoolean(if_missing=False)
         overwrite_default_group = v.StringBoolean(if_missing=False)
         overwrite_default_user_group = v.StringBoolean(if_missing=False)
         default_repo_perm = v.OneOf(repo_perms_choices)
         default_group_perm = v.OneOf(group_perms_choices)
         default_user_group_perm = v.OneOf(user_group_perms_choices)
 
     return _ObjectPermissionsForm
 
 
 def UserPermissionsForm(create_choices, create_on_write_choices,
                         repo_group_create_choices, user_group_create_choices,
                         fork_choices, inherit_default_permissions_choices):
     class _DefaultPermissionsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         anonymous = v.StringBoolean(if_missing=False)
 
         default_repo_create = v.OneOf(create_choices)
         default_repo_create_on_write = v.OneOf(create_on_write_choices)
         default_user_group_create = v.OneOf(user_group_create_choices)
         default_repo_group_create = v.OneOf(repo_group_create_choices)
         default_fork_create = v.OneOf(fork_choices)
         default_inherit_default_permissions = v.OneOf(inherit_default_permissions_choices)
 
     return _DefaultPermissionsForm
 
 
 def UserIndividualPermissionsForm():
     class _DefaultPermissionsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         inherit_default_permissions = v.StringBoolean(if_missing=False)
 
     return _DefaultPermissionsForm
 
 
 def DefaultsForm(edit=False, old_data={}, supported_backends=BACKENDS.keys()):
     class _DefaultsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         default_repo_type = v.OneOf(supported_backends)
         default_repo_private = v.StringBoolean(if_missing=False)
         default_repo_enable_statistics = v.StringBoolean(if_missing=False)
         default_repo_enable_downloads = v.StringBoolean(if_missing=False)
         default_repo_enable_locking = v.StringBoolean(if_missing=False)
 
     return _DefaultsForm
 
 
 def AuthSettingsForm():
     class _AuthSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         auth_plugins = All(v.ValidAuthPlugins(),
                            v.UniqueListFromString()(not_empty=True))
 
     return _AuthSettingsForm
 
 
 def UserExtraEmailForm():
     class _UserExtraEmailForm(formencode.Schema):
         email = All(v.UniqSystemEmail(), v.Email(not_empty=True))
     return _UserExtraEmailForm
 
 
 def UserExtraIpForm():
     class _UserExtraIpForm(formencode.Schema):
         ip = v.ValidIp()(not_empty=True)
     return _UserExtraIpForm
 
 
 def PullRequestForm(repo_id):
     class _PullRequestForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         user = v.UnicodeString(strip=True, required=True)
         source_repo = v.UnicodeString(strip=True, required=True)
         source_ref = v.UnicodeString(strip=True, required=True)
         target_repo = v.UnicodeString(strip=True, required=True)
         target_ref = v.UnicodeString(strip=True, required=True)
         revisions = All(#v.NotReviewedRevisions(repo_id)(),
                         v.UniqueList()(not_empty=True))
         review_members = v.UniqueList(convert=int)(not_empty=True)
 
         pullrequest_title = v.UnicodeString(strip=True, required=True)
         pullrequest_desc = v.UnicodeString(strip=True, required=False)
 
     return _PullRequestForm
 
 
 def IssueTrackerPatternsForm():
     class _IssueTrackerPatternsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         chained_validators = [v.ValidPattern()]
     return _IssueTrackerPatternsForm

rhodecode/model/settings.py

@@ -1,716 +1,730 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  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 hashlib
 import logging
 from collections import namedtuple
 from functools import wraps
 
 from rhodecode.lib import caches
-from rhodecode.lib.caching_query import FromCache
 from rhodecode.lib.utils2 import (
     Optional, AttributeDict, safe_str, remove_prefix, str2bool)
+from rhodecode.lib.vcs.backends import base
 from rhodecode.model import BaseModel
 from rhodecode.model.db import (
     RepoRhodeCodeUi, RepoRhodeCodeSetting, RhodeCodeUi, RhodeCodeSetting)
 from rhodecode.model.meta import Session
 
 
 log = logging.getLogger(__name__)
 
 
 UiSetting = namedtuple(
     'UiSetting', ['section', 'key', 'value', 'active'])
 
 SOCIAL_PLUGINS_LIST = ['github', 'bitbucket', 'twitter', 'google']
 
 
 class SettingNotFound(Exception):
     def __init__(self):
         super(SettingNotFound, self).__init__('Setting is not found')
 
 
 class SettingsModel(BaseModel):
     BUILTIN_HOOKS = (
         RhodeCodeUi.HOOK_REPO_SIZE, RhodeCodeUi.HOOK_PUSH,
         RhodeCodeUi.HOOK_PRE_PUSH, RhodeCodeUi.HOOK_PULL,
         RhodeCodeUi.HOOK_PRE_PULL)
     HOOKS_SECTION = 'hooks'
 
     def __init__(self, sa=None, repo=None):
         self.repo = repo
         self.UiDbModel = RepoRhodeCodeUi if repo else RhodeCodeUi
         self.SettingsDbModel = (
             RepoRhodeCodeSetting if repo else RhodeCodeSetting)
         super(SettingsModel, self).__init__(sa)
 
     def get_ui_by_key(self, key):
         q = self.UiDbModel.query()
         q = q.filter(self.UiDbModel.ui_key == key)
         q = self._filter_by_repo(RepoRhodeCodeUi, q)
         return q.scalar()
 
     def get_ui_by_section(self, section):
         q = self.UiDbModel.query()
         q = q.filter(self.UiDbModel.ui_section == section)
         q = self._filter_by_repo(RepoRhodeCodeUi, q)
         return q.all()
 
     def get_ui_by_section_and_key(self, section, key):
         q = self.UiDbModel.query()
         q = q.filter(self.UiDbModel.ui_section == section)
         q = q.filter(self.UiDbModel.ui_key == key)
         q = self._filter_by_repo(RepoRhodeCodeUi, q)
         return q.scalar()
 
     def get_ui(self, section=None, key=None):
         q = self.UiDbModel.query()
         q = self._filter_by_repo(RepoRhodeCodeUi, q)
 
         if section:
             q = q.filter(self.UiDbModel.ui_section == section)
         if key:
             q = q.filter(self.UiDbModel.ui_key == key)
 
         # TODO: mikhail: add caching
         result = [
             UiSetting(
                 section=safe_str(r.ui_section), key=safe_str(r.ui_key),
                 value=safe_str(r.ui_value), active=r.ui_active
             )
             for r in q.all()
         ]
         return result
 
     def get_builtin_hooks(self):
         q = self.UiDbModel.query()
         q = q.filter(self.UiDbModel.ui_key.in_(self.BUILTIN_HOOKS))
         return self._get_hooks(q)
 
     def get_custom_hooks(self):
         q = self.UiDbModel.query()
         q = q.filter(~self.UiDbModel.ui_key.in_(self.BUILTIN_HOOKS))
         return self._get_hooks(q)
 
     def create_ui_section_value(self, section, val, key=None, active=True):
         new_ui = self.UiDbModel()
         new_ui.ui_section = section
         new_ui.ui_value = val
         new_ui.ui_active = active
 
         if self.repo:
             repo = self._get_repo(self.repo)
             repository_id = repo.repo_id
             new_ui.repository_id = repository_id
 
         if not key:
             # keys are unique so they need appended info
             if self.repo:
                 key = hashlib.sha1(
                     '{}{}{}'.format(section, val, repository_id)).hexdigest()
             else:
                 key = hashlib.sha1('{}{}'.format(section, val)).hexdigest()
 
         new_ui.ui_key = key
 
         Session().add(new_ui)
         return new_ui
 
     def create_or_update_hook(self, key, value):
         ui = (
             self.get_ui_by_section_and_key(self.HOOKS_SECTION, key) or
             self.UiDbModel())
         ui.ui_section = self.HOOKS_SECTION
         ui.ui_active = True
         ui.ui_key = key
         ui.ui_value = value
 
         if self.repo:
             repo = self._get_repo(self.repo)
             repository_id = repo.repo_id
             ui.repository_id = repository_id
 
         Session().add(ui)
         return ui
 
     def delete_ui(self, id_):
         ui = self.UiDbModel.get(id_)
         if not ui:
             raise SettingNotFound()
         Session().delete(ui)
 
     def get_setting_by_name(self, name):
         q = self._get_settings_query()
         q = q.filter(self.SettingsDbModel.app_settings_name == name)
         return q.scalar()
 
     def create_or_update_setting(
             self, name, val=Optional(''), type_=Optional('unicode')):
         """
         Creates or updates RhodeCode setting. If updates is triggered it will
         only update parameters that are explicityl set Optional instance will
         be skipped
 
         :param name:
         :param val:
         :param type_:
         :return:
         """
 
         res = self.get_setting_by_name(name)
         repo = self._get_repo(self.repo) if self.repo else None
 
         if not res:
             val = Optional.extract(val)
             type_ = Optional.extract(type_)
 
             args = (
                 (repo.repo_id, name, val, type_)
                 if repo else (name, val, type_))
             res = self.SettingsDbModel(*args)
 
         else:
             if self.repo:
                 res.repository_id = repo.repo_id
 
             res.app_settings_name = name
             if not isinstance(type_, Optional):
                 # update if set
                 res.app_settings_type = type_
             if not isinstance(val, Optional):
                 # update if set
                 res.app_settings_value = val
 
         Session().add(res)
         return res
 
     def invalidate_settings_cache(self):
         namespace = 'rhodecode_settings'
         cache_manager = caches.get_cache_manager('sql_cache_short', namespace)
         caches.clear_cache_manager(cache_manager)
 
     def get_all_settings(self, cache=False):
         def _compute():
             q = self._get_settings_query()
             if not q:
                 raise Exception('Could not get application settings !')
 
             settings = {
                 'rhodecode_' + result.app_settings_name: result.app_settings_value
                 for result in q
             }
             return settings
 
         if cache:
             log.debug('Fetching app settings using cache')
             repo = self._get_repo(self.repo) if self.repo else None
             namespace = 'rhodecode_settings'
             cache_manager = caches.get_cache_manager(
                 'sql_cache_short', namespace)
             _cache_key = (
                 "get_repo_{}_settings".format(repo.repo_id)
                 if repo else "get_app_settings")
 
             return cache_manager.get(_cache_key, createfunc=_compute)
 
         else:
             return _compute()
 
     def get_auth_settings(self):
         q = self._get_settings_query()
         q = q.filter(
             self.SettingsDbModel.app_settings_name.startswith('auth_'))
         rows = q.all()
         auth_settings = {
             row.app_settings_name: row.app_settings_value for row in rows}
         return auth_settings
 
     def get_auth_plugins(self):
         auth_plugins = self.get_setting_by_name("auth_plugins")
         return auth_plugins.app_settings_value
 
     def get_default_repo_settings(self, strip_prefix=False):
         q = self._get_settings_query()
         q = q.filter(
             self.SettingsDbModel.app_settings_name.startswith('default_'))
         rows = q.all()
 
         result = {}
         for row in rows:
             key = row.app_settings_name
             if strip_prefix:
                 key = remove_prefix(key, prefix='default_')
             result.update({key: row.app_settings_value})
         return result
 
     def get_repo(self):
         repo = self._get_repo(self.repo)
         if not repo:
             raise Exception(
                 'Repository {} cannot be found'.format(self.repo))
         return repo
 
     def _filter_by_repo(self, model, query):
         if self.repo:
             repo = self.get_repo()
             query = query.filter(model.repository_id == repo.repo_id)
         return query
 
     def _get_hooks(self, query):
         query = query.filter(self.UiDbModel.ui_section == self.HOOKS_SECTION)
         query = self._filter_by_repo(RepoRhodeCodeUi, query)
         return query.all()
 
     def _get_settings_query(self):
         q = self.SettingsDbModel.query()
         return self._filter_by_repo(RepoRhodeCodeSetting, q)
 
     def list_enabled_social_plugins(self, settings):
         enabled = []
         for plug in SOCIAL_PLUGINS_LIST:
             if str2bool(settings.get('rhodecode_auth_{}_enabled'.format(plug)
                                      )):
                 enabled.append(plug)
         return enabled
 
 
 def assert_repo_settings(func):
     @wraps(func)
     def _wrapper(self, *args, **kwargs):
         if not self.repo_settings:
             raise Exception('Repository is not specified')
         return func(self, *args, **kwargs)
     return _wrapper
 
 
 class IssueTrackerSettingsModel(object):
     INHERIT_SETTINGS = 'inherit_issue_tracker_settings'
     SETTINGS_PREFIX = 'issuetracker_'
 
     def __init__(self, sa=None, repo=None):
         self.global_settings = SettingsModel(sa=sa)
         self.repo_settings = SettingsModel(sa=sa, repo=repo) if repo else None
 
     @property
     def inherit_global_settings(self):
         if not self.repo_settings:
             return True
         setting = self.repo_settings.get_setting_by_name(self.INHERIT_SETTINGS)
         return setting.app_settings_value if setting else True
 
     @inherit_global_settings.setter
     def inherit_global_settings(self, value):
         if self.repo_settings:
             settings = self.repo_settings.create_or_update_setting(
                 self.INHERIT_SETTINGS, value, type_='bool')
             Session().add(settings)
 
     def _get_keyname(self, key, uid, prefix=''):
         return '{0}{1}{2}_{3}'.format(
             prefix, self.SETTINGS_PREFIX, key, uid)
 
     def _make_dict_for_settings(self, qs):
         prefix_match = self._get_keyname('pat', '', 'rhodecode_')
 
         issuetracker_entries = {}
         # create keys
         for k, v in qs.items():
             if k.startswith(prefix_match):
                 uid = k[len(prefix_match):]
                 issuetracker_entries[uid] = None
 
         # populate
         for uid in issuetracker_entries:
             issuetracker_entries[uid] = AttributeDict({
                 'pat': qs.get(self._get_keyname('pat', uid, 'rhodecode_')),
                 'url': qs.get(self._get_keyname('url', uid, 'rhodecode_')),
                 'pref': qs.get(self._get_keyname('pref', uid, 'rhodecode_')),
                 'desc': qs.get(self._get_keyname('desc', uid, 'rhodecode_')),
             })
         return issuetracker_entries
 
     def get_global_settings(self, cache=False):
         """
         Returns list of global issue tracker settings
         """
         defaults = self.global_settings.get_all_settings(cache=cache)
         settings = self._make_dict_for_settings(defaults)
         return settings
 
     def get_repo_settings(self, cache=False):
         """
         Returns list of issue tracker settings per repository
         """
         if not self.repo_settings:
             raise Exception('Repository is not specified')
         all_settings = self.repo_settings.get_all_settings(cache=cache)
         settings = self._make_dict_for_settings(all_settings)
         return settings
 
     def get_settings(self, cache=False):
         if self.inherit_global_settings:
             return self.get_global_settings(cache=cache)
         else:
             return self.get_repo_settings(cache=cache)
 
     def delete_entries(self, uid):
         if self.repo_settings:
             all_patterns = self.get_repo_settings()
             settings_model = self.repo_settings
         else:
             all_patterns = self.get_global_settings()
             settings_model = self.global_settings
         entries = all_patterns.get(uid)
 
         for del_key in entries:
             setting_name = self._get_keyname(del_key, uid)
             entry = settings_model.get_setting_by_name(setting_name)
             if entry:
                 Session().delete(entry)
 
         Session().commit()
 
     def create_or_update_setting(
             self, name, val=Optional(''), type_=Optional('unicode')):
         if self.repo_settings:
             setting = self.repo_settings.create_or_update_setting(
                 name, val, type_)
         else:
             setting = self.global_settings.create_or_update_setting(
                 name, val, type_)
         return setting
 
 
 class VcsSettingsModel(object):
 
     INHERIT_SETTINGS = 'inherit_vcs_settings'
     GENERAL_SETTINGS = (
-        'use_outdated_comments', 'pr_merge_enabled',
+        'use_outdated_comments',
+        'pr_merge_enabled',
         'hg_use_rebase_for_merging')
+
     HOOKS_SETTINGS = (
         ('hooks', 'changegroup.repo_size'),
         ('hooks', 'changegroup.push_logger'),
         ('hooks', 'outgoing.pull_logger'))
     HG_SETTINGS = (
-        ('extensions', 'largefiles'), ('phases', 'publish'))
-    GLOBAL_HG_SETTINGS = HG_SETTINGS + (('extensions', 'hgsubversion'), )
+        ('extensions', 'largefiles'),
+        ('phases', 'publish'))
+    GLOBAL_HG_SETTINGS = (
+        ('extensions', 'largefiles'),
+        ('phases', 'publish'),
+        ('extensions', 'hgsubversion'))
     GLOBAL_SVN_SETTINGS = (
-        'rhodecode_proxy_subversion_http_requests',
-        'rhodecode_subversion_http_server_url')
+        ('vcs_svn_proxy', 'http_requests_enabled'),
+        ('vcs_svn_proxy', 'http_server_url'))
+
     SVN_BRANCH_SECTION = 'vcs_svn_branch'
     SVN_TAG_SECTION = 'vcs_svn_tag'
     SSL_SETTING = ('web', 'push_ssl')
     PATH_SETTING = ('paths', '/')
 
     def __init__(self, sa=None, repo=None):
         self.global_settings = SettingsModel(sa=sa)
         self.repo_settings = SettingsModel(sa=sa, repo=repo) if repo else None
         self._ui_settings = self.HG_SETTINGS + self.HOOKS_SETTINGS
         self._svn_sections = (self.SVN_BRANCH_SECTION, self.SVN_TAG_SECTION)
 
     @property
     @assert_repo_settings
     def inherit_global_settings(self):
         setting = self.repo_settings.get_setting_by_name(self.INHERIT_SETTINGS)
         return setting.app_settings_value if setting else True
 
     @inherit_global_settings.setter
     @assert_repo_settings
     def inherit_global_settings(self, value):
         self.repo_settings.create_or_update_setting(
             self.INHERIT_SETTINGS, value, type_='bool')
 
     def get_global_svn_branch_patterns(self):
         return self.global_settings.get_ui_by_section(self.SVN_BRANCH_SECTION)
 
     @assert_repo_settings
     def get_repo_svn_branch_patterns(self):
         return self.repo_settings.get_ui_by_section(self.SVN_BRANCH_SECTION)
 
     def get_global_svn_tag_patterns(self):
         return self.global_settings.get_ui_by_section(self.SVN_TAG_SECTION)
 
     @assert_repo_settings
     def get_repo_svn_tag_patterns(self):
         return self.repo_settings.get_ui_by_section(self.SVN_TAG_SECTION)
 
     def get_global_settings(self):
         return self._collect_all_settings(global_=True)
 
     @assert_repo_settings
     def get_repo_settings(self):
         return self._collect_all_settings(global_=False)
 
     @assert_repo_settings
     def create_or_update_repo_settings(
             self, data, inherit_global_settings=False):
         from rhodecode.model.scm import ScmModel
 
         self.inherit_global_settings = inherit_global_settings
 
         repo = self.repo_settings.get_repo()
         if not inherit_global_settings:
             if repo.repo_type == 'svn':
                 self.create_repo_svn_settings(data)
             else:
                 self.create_or_update_repo_hook_settings(data)
                 self.create_or_update_repo_pr_settings(data)
 
             if repo.repo_type == 'hg':
                 self.create_or_update_repo_hg_settings(data)
 
         ScmModel().mark_for_invalidation(repo.repo_name, delete=True)
 
     @assert_repo_settings
     def create_or_update_repo_hook_settings(self, data):
         for section, key in self.HOOKS_SETTINGS:
             data_key = self._get_form_ui_key(section, key)
             if data_key not in data:
                 raise ValueError(
                     'The given data does not contain {} key'.format(data_key))
 
             active = data.get(data_key)
             repo_setting = self.repo_settings.get_ui_by_section_and_key(
                 section, key)
             if not repo_setting:
                 global_setting = self.global_settings.\
                     get_ui_by_section_and_key(section, key)
                 self.repo_settings.create_ui_section_value(
                     section, global_setting.ui_value, key=key, active=active)
             else:
                 repo_setting.ui_active = active
                 Session().add(repo_setting)
 
     def update_global_hook_settings(self, data):
         for section, key in self.HOOKS_SETTINGS:
             data_key = self._get_form_ui_key(section, key)
             if data_key not in data:
                 raise ValueError(
                     'The given data does not contain {} key'.format(data_key))
             active = data.get(data_key)
             repo_setting = self.global_settings.get_ui_by_section_and_key(
                 section, key)
             repo_setting.ui_active = active
             Session().add(repo_setting)
 
     @assert_repo_settings
     def create_or_update_repo_pr_settings(self, data):
         return self._create_or_update_general_settings(
             self.repo_settings, data)
 
     def create_or_update_global_pr_settings(self, data):
         return self._create_or_update_general_settings(
             self.global_settings, data)
 
     @assert_repo_settings
     def create_repo_svn_settings(self, data):
         return self._create_svn_settings(self.repo_settings, data)
 
-    def create_global_svn_settings(self, data):
-        return self._create_svn_settings(self.global_settings, data)
-
     @assert_repo_settings
     def create_or_update_repo_hg_settings(self, data):
         largefiles, phases = self.HG_SETTINGS
-        largefiles_key, phases_key = self._get_hg_settings(
+        largefiles_key, phases_key = self._get_settings_keys(
             self.HG_SETTINGS, data)
         self._create_or_update_ui(
             self.repo_settings, *largefiles, value='',
             active=data[largefiles_key])
         self._create_or_update_ui(
             self.repo_settings, *phases, value=safe_str(data[phases_key]))
 
     def create_or_update_global_hg_settings(self, data):
-        largefiles, phases, subversion = self.GLOBAL_HG_SETTINGS
-        largefiles_key, phases_key, subversion_key = self._get_hg_settings(
+        largefiles, phases, hgsubversion = self.GLOBAL_HG_SETTINGS
+        largefiles_key, phases_key, subversion_key = self._get_settings_keys(
             self.GLOBAL_HG_SETTINGS, data)
         self._create_or_update_ui(
             self.global_settings, *largefiles, value='',
             active=data[largefiles_key])
         self._create_or_update_ui(
             self.global_settings, *phases, value=safe_str(data[phases_key]))
         self._create_or_update_ui(
-            self.global_settings, *subversion, active=data[subversion_key])
+            self.global_settings, *hgsubversion, active=data[subversion_key])
 
     def create_or_update_global_svn_settings(self, data):
-        rhodecode_proxy_subversion_http_requests,
-        rhodecode_subversion_http_server_url = self.GLOBAL_SVN_SETTINGS
-        rhodecode_proxy_subversion_http_requests_key,
-        rhodecode_subversion_http_server_url_key = self._get_svn_settings(
+        # branch/tags patterns
+        self._create_svn_settings(self.global_settings, data)
+
+        http_requests_enabled, http_server_url = self.GLOBAL_SVN_SETTINGS
+        http_requests_enabled_key, http_server_url_key = self._get_settings_keys(
             self.GLOBAL_SVN_SETTINGS, data)
+
         self._create_or_update_ui(
-            self.global_settings,
-            *rhodecode_proxy_subversion_http_requests,
-            value=safe_str(data[rhodecode_proxy_subversion_http_requests_key]))
+            self.global_settings, *http_requests_enabled,
+            value=safe_str(data[http_requests_enabled_key]))
         self._create_or_update_ui(
-            self.global_settings,
-            *rhodecode_subversion_http_server_url,
-            active=data[rhodecode_subversion_http_server_url_key])
+            self.global_settings, *http_server_url,
+            value=data[http_server_url_key])
 
     def update_global_ssl_setting(self, value):
         self._create_or_update_ui(
             self.global_settings, *self.SSL_SETTING, value=value)
 
     def update_global_path_setting(self, value):
         self._create_or_update_ui(
             self.global_settings, *self.PATH_SETTING, value=value)
 
     @assert_repo_settings
     def delete_repo_svn_pattern(self, id_):
         self.repo_settings.delete_ui(id_)
 
     def delete_global_svn_pattern(self, id_):
         self.global_settings.delete_ui(id_)
 
     @assert_repo_settings
     def get_repo_ui_settings(self, section=None, key=None):
         global_uis = self.global_settings.get_ui(section, key)
         repo_uis = self.repo_settings.get_ui(section, key)
         filtered_repo_uis = self._filter_ui_settings(repo_uis)
         filtered_repo_uis_keys = [
             (s.section, s.key) for s in filtered_repo_uis]
 
         def _is_global_ui_filtered(ui):
             return (
                 (ui.section, ui.key) in filtered_repo_uis_keys
                 or ui.section in self._svn_sections)
 
         filtered_global_uis = [
             ui for ui in global_uis if not _is_global_ui_filtered(ui)]
 
         return filtered_global_uis + filtered_repo_uis
 
     def get_global_ui_settings(self, section=None, key=None):
         return self.global_settings.get_ui(section, key)
 
+    def get_ui_settings_as_config_obj(self, section=None, key=None):
+        config = base.Config()
+
+        ui_settings = self.get_ui_settings(section=section, key=key)
+
+        for entry in ui_settings:
+            config.set(entry.section, entry.key, entry.value)
+
+        return config
+
     def get_ui_settings(self, section=None, key=None):
         if not self.repo_settings or self.inherit_global_settings:
             return self.get_global_ui_settings(section, key)
         else:
             return self.get_repo_ui_settings(section, key)
 
     def get_svn_patterns(self, section=None):
         if not self.repo_settings:
             return self.get_global_ui_settings(section)
         else:
             return self.get_repo_ui_settings(section)
 
     @assert_repo_settings
     def get_repo_general_settings(self):
         global_settings = self.global_settings.get_all_settings()
         repo_settings = self.repo_settings.get_all_settings()
         filtered_repo_settings = self._filter_general_settings(repo_settings)
         global_settings.update(filtered_repo_settings)
         return global_settings
 
     def get_global_general_settings(self):
         return self.global_settings.get_all_settings()
 
     def get_general_settings(self):
         if not self.repo_settings or self.inherit_global_settings:
             return self.get_global_general_settings()
         else:
             return self.get_repo_general_settings()
 
     def get_repos_location(self):
         return self.global_settings.get_ui_by_key('/').ui_value
 
     def _filter_ui_settings(self, settings):
         filtered_settings = [
             s for s in settings if self._should_keep_setting(s)]
         return filtered_settings
 
     def _should_keep_setting(self, setting):
         keep = (
             (setting.section, setting.key) in self._ui_settings or
             setting.section in self._svn_sections)
         return keep
 
     def _filter_general_settings(self, settings):
         keys = ['rhodecode_{}'.format(key) for key in self.GENERAL_SETTINGS]
         return {
             k: settings[k]
             for k in settings if k in keys}
 
     def _collect_all_settings(self, global_=False):
         settings = self.global_settings if global_ else self.repo_settings
         result = {}
 
         for section, key in self._ui_settings:
             ui = settings.get_ui_by_section_and_key(section, key)
             result_key = self._get_form_ui_key(section, key)
             if ui:
                 if section in ('hooks', 'extensions'):
                     result[result_key] = ui.ui_active
                 else:
                     result[result_key] = ui.ui_value
 
         for name in self.GENERAL_SETTINGS:
             setting = settings.get_setting_by_name(name)
             if setting:
                 result_key = 'rhodecode_{}'.format(name)
                 result[result_key] = setting.app_settings_value
 
         return result
 
     def _get_form_ui_key(self, section, key):
         return '{section}_{key}'.format(
             section=section, key=key.replace('.', '_'))
 
     def _create_or_update_ui(
             self, settings, section, key, value=None, active=None):
         ui = settings.get_ui_by_section_and_key(section, key)
         if not ui:
             active = True if active is None else active
             settings.create_ui_section_value(
                 section, value, key=key, active=active)
         else:
             if active is not None:
                 ui.ui_active = active
             if value is not None:
                 ui.ui_value = value
             Session().add(ui)
 
     def _create_svn_settings(self, settings, data):
         svn_settings = {
             'new_svn_branch': self.SVN_BRANCH_SECTION,
             'new_svn_tag': self.SVN_TAG_SECTION
         }
         for key in svn_settings:
             if data.get(key):
                 settings.create_ui_section_value(svn_settings[key], data[key])
 
     def _create_or_update_general_settings(self, settings, data):
         for name in self.GENERAL_SETTINGS:
             data_key = 'rhodecode_{}'.format(name)
             if data_key not in data:
                 raise ValueError(
                     'The given data does not contain {} key'.format(data_key))
             setting = settings.create_or_update_setting(
                 name, data[data_key], 'bool')
             Session().add(setting)
 
-    def _get_hg_settings(self, settings, data):
+    def _get_settings_keys(self, settings, data):
         data_keys = [self._get_form_ui_key(*s) for s in settings]
         for data_key in data_keys:
             if data_key not in data:
                 raise ValueError(
                     'The given data does not contain {} key'.format(data_key))
         return data_keys

rhodecode/templates/base/vcs_settings.html

@@ -1,259 +1,267 b''
 ## snippet for displaying vcs settings
 ## usage:
 ##    <%namespace name="vcss" file="/base/vcssettings.html"/>
 ##    ${vcss.vcs_settings_fields()}
 
 <%def name="vcs_settings_fields(suffix='', svn_branch_patterns=None, svn_tag_patterns=None, repo_type=None, display_globals=False, allow_repo_location_change=False, **kwargs)">
     % if display_globals:
         <div class="panel panel-default">
             <div class="panel-heading" id="general">
                 <h3 class="panel-title">${_('General')}</h3>
             </div>
             <div class="panel-body">
                 <div class="field">
                     <div class="checkbox">
                         ${h.checkbox('web_push_ssl' + suffix, 'True')}
                         <label for="web_push_ssl${suffix}">${_('Require SSL for vcs operations')}</label>
                     </div>
                     <div class="label">
                         <span class="help-block">${_('Activate to set RhodeCode to require SSL for pushing or pulling. If SSL certificate is missing it will return a HTTP Error 406: Not Acceptable.')}</span>
                     </div>
                  </div>
             </div>
         </div>
     % endif
 
     % if display_globals:
         <div class="panel panel-default">
             <div class="panel-heading">
                 <h3 class="panel-title">${_('Main Storage Location')}</h3>
             </div>
             <div class="panel-body">
                 <div class="field">
                     <div class="inputx locked_input">
                         %if allow_repo_location_change:
                             ${h.text('paths_root_path',size=59,readonly="readonly", class_="disabled")}
                             <span id="path_unlock" class="tooltip"
                                     title="${h.tooltip(_('Click to unlock. You must restart RhodeCode in order to make this setting take effect.'))}">
                                 <div class="btn btn-default lock_input_button"><i id="path_unlock_icon" class="icon-lock"></i></div>
                             </span>
                         %else:
                             ${_('Repository location change is disabled. You can enable this by changing the `allow_repo_location_change` inside .ini file.')}
                             ## form still requires this but we cannot internally change it anyway
                             ${h.hidden('paths_root_path',size=30,readonly="readonly", class_="disabled")}
                         %endif
                     </div>
                 </div>
                 <div class="label">
                     <span class="help-block">${_('Filesystem location where repositories should be stored. After changing this value a restart and rescan of the repository folder are required.')}</span>
                 </div>
             </div>
         </div>
     % endif
 
     % if display_globals or repo_type in ['git', 'hg']:
         <div class="panel panel-default">
             <div class="panel-heading" id="general">
                 <h3 class="panel-title">${_('Internal Hooks')}</h3>
             </div>
             <div class="panel-body">
                 <div class="field">
                     <div class="checkbox">
                         ${h.checkbox('hooks_changegroup_repo_size' + suffix, 'True', **kwargs)}
                         <label for="hooks_changegroup_repo_size${suffix}">${_('Show repository size after push')}</label>
                     </div>
 
                     <div class="label">
                         <span class="help-block">${_('Trigger a hook that calculates repository size after each push.')}</span>
                     </div>
                     <div class="checkbox">
                         ${h.checkbox('hooks_changegroup_push_logger' + suffix, 'True', **kwargs)}
                         <label for="hooks_changegroup_push_logger${suffix}">${_('Execute pre/post push hooks')}</label>
                     </div>
                     <div class="label">
                         <span class="help-block">${_('Execute Built in pre/post push hooks. This also executes rcextensions hooks.')}</span>
                     </div>
                     <div class="checkbox">
                         ${h.checkbox('hooks_outgoing_pull_logger' + suffix, 'True', **kwargs)}
                         <label for="hooks_outgoing_pull_logger${suffix}">${_('Execute pre/post pull hooks')}</label>
                     </div>
                     <div class="label">
                         <span class="help-block">${_('Execute Built in pre/post pull hooks. This also executes rcextensions hooks.')}</span>
                     </div>
                 </div>
             </div>
         </div>
     % endif
 
     % if display_globals or repo_type in ['hg']:
         <div class="panel panel-default">
             <div class="panel-heading">
                 <h3 class="panel-title">${_('Mercurial Settings')}</h3>
             </div>
             <div class="panel-body">
                 <div class="checkbox">
                     ${h.checkbox('extensions_largefiles' + suffix, 'True', **kwargs)}
                     <label for="extensions_largefiles${suffix}">${_('Enable largefiles extension')}</label>
                 </div>
                 <div class="label">
                     <span class="help-block">${_('Enable Largefiles extensions for all repositories.')}</span>
                 </div>
                 <div class="checkbox">
                     ${h.checkbox('phases_publish' + suffix, 'True', **kwargs)}
                     <label for="phases_publish${suffix}">${_('Set repositories as publishing') if display_globals else _('Set repository as publishing')}</label>
                 </div>
                 <div class="label">
                     <span class="help-block">${_('When this is enabled all commits in the repository are seen as public commits by clients.')}</span>
                 </div>
                 % if display_globals:
                     <div class="checkbox">
                         ${h.checkbox('extensions_hgsubversion' + suffix,'True')}
                         <label for="extensions_hgsubversion${suffix}">${_('Enable hgsubversion extension')}</label>
                     </div>
                     <div class="label">
                         <span class="help-block">${_('Requires hgsubversion library to be installed. Allows cloning remote SVN repositories and migrates them to Mercurial type.')}</span>
                     </div>
                 % endif
             </div>
         </div>
         ## LABS for HG
         % if c.labs_active:
             <div class="panel panel-danger">
                 <div class="panel-heading">
                     <h3 class="panel-title">${_('Mercurial Labs Settings')} (${_('These features are considered experimental and may not work as expected.')})</h3>
                 </div>
                 <div class="panel-body">
 
                     <div class="checkbox">
                         ${h.checkbox('rhodecode_hg_use_rebase_for_merging' + suffix, 'True', **kwargs)}
                         <label for="rhodecode_hg_use_rebase_for_merging{suffix}">${_('Use rebase as merge strategy')}</label>
                     </div>
                     <div class="label">
                         <span class="help-block">${_('Use rebase instead of creating a merge commit when merging via web interface.')}</span>
                     </div>
 
                 </div>
             </div>
         % endif
 
     % endif
 
-    % if display_globals or repo_type in ['svn']:
+    % if display_globals:
         <div class="panel panel-default">
             <div class="panel-heading">
-                <h3 class="panel-title">${_('Subversion Settings')}</h3>
+                <h3 class="panel-title">${_('Global Subversion Settings')}</h3>
             </div>
             <div class="panel-body">
                 <div class="field">
                     <div class="checkbox">
-                        ${h.checkbox('rhodecode_proxy_subversion_http_requests' + suffix, 'True', **kwargs)}
-                        <label for="rhodecode_proxy_subversion_http_requests${suffix}">${_('Proxy subversion HTTP requests')}</label>
+                        ${h.checkbox('vcs_svn_proxy_http_requests_enabled' + suffix, 'True', **kwargs)}
+                        <label for="vcs_svn_proxy_http_requests_enabled{suffix}">${_('Proxy subversion HTTP requests')}</label>
                     </div>
                     <div class="label">
                         <span class="help-block">${_('Subversion HTTP Support. Enables communication with SVN over HTTP protocol.')}</span>
                     </div>
                 </div>
                 <div class="field">
                     <div class="label">
-                        <label for="rhodecode_subversion_http_server_url">${_('Subversion HTTP Server URL')}</label><br/>
+                        <label for="vcs_svn_proxy_http_server_url">${_('Subversion HTTP Server URL')}</label><br/>
                     </div>
                     <div class="input">
-                        ${h.text('rhodecode_subversion_http_server_url',size=59)}
+                        ${h.text('vcs_svn_proxy_http_server_url',size=59)}
+                    </div>
+                </div>
                     </div>
                 </div>
-                <div class="field">
-                    <div class="label">
-                        <span class="help-block">${_('Url to Apache Proxy, e.g. http://localhost:8080/')}</span>
+    % endif
+
+    % if display_globals or repo_type in ['svn']:
+        <div class="panel panel-default">
+            <div class="panel-heading">
+                <h3 class="panel-title">${_('Subversion Settings')}</h3>
                     </div>
-                 </div>
+            <div class="panel-body">
                 <div class="field">
                     <div class="content" >
                         <label>${_('Repository patterns')}</label><br/>
                     </div>
                 </div>
                 <div class="label">
                     <span class="help-block">${_('Patterns for identifying SVN branches and tags. For recursive search, use "*". Eg.: "/branches/*"')}</span>
                 </div>
 
                 <div class="field branch_patterns">
                     <div class="input" >
                         <label>${_('Branches')}:</label><br/>
                     </div>
                     % if svn_branch_patterns:
                         % for branch in svn_branch_patterns:
                         <div class="input adjacent"   id="${'id%s' % branch.ui_id}">
                             ${h.hidden('branch_ui_key' + suffix, branch.ui_key)}
                             ${h.text('branch_value_%d' % branch.ui_id + suffix, branch.ui_value, size=59, readonly="readonly", class_='disabled')}
                             % if kwargs.get('disabled') != 'disabled':
                                 <span class="btn btn-x" onclick="ajaxDeletePattern(${branch.ui_id},'${'id%s' % branch.ui_id}')">
                                     ${_('Delete')}
                                 </span>
                             % endif
                         </div>
                         % endfor
                     %endif
                 </div>
                 % if kwargs.get('disabled') != 'disabled':
                     <div class="field branch_patterns">
                         <div class="input" >
                             ${h.text('new_svn_branch',size=59,placeholder='New branch pattern')}
                         </div>
                     </div>
                 % endif
                 <div class="field tag_patterns">
                     <div class="input" >
                         <label>${_('Tags')}:</label><br/>
                     </div>
                     % if svn_tag_patterns:
                         % for tag in svn_tag_patterns:
                         <div class="input"  id="${'id%s' % tag.ui_id + suffix}">
                             ${h.hidden('tag_ui_key' + suffix, tag.ui_key)}
                             ${h.text('tag_ui_value_new_%d' % tag.ui_id + suffix, tag.ui_value, size=59, readonly="readonly", class_='disabled tag_input')}
                             % if kwargs.get('disabled') != 'disabled':
                                 <span class="btn btn-x" onclick="ajaxDeletePattern(${tag.ui_id},'${'id%s' % tag.ui_id}')">
                                     ${_('Delete')}
                                 </span>
                             %endif
                         </div>
                         % endfor
                     % endif
                 </div>
                 % if kwargs.get('disabled') != 'disabled':
                     <div class="field tag_patterns">
                         <div class="input" >
                             ${h.text('new_svn_tag' + suffix, size=59, placeholder='New tag pattern')}
                         </div>
                     </div>
                 %endif
             </div>
         </div>
     % else:
         ${h.hidden('new_svn_branch' + suffix, '')}
         ${h.hidden('new_svn_tag' + suffix, '')}
     % endif
 
+
+
+
     % if display_globals or repo_type in ['hg', 'git']:
         <div class="panel panel-default">
             <div class="panel-heading">
                 <h3 class="panel-title">${_('Pull Request Settings')}</h3>
             </div>
             <div class="panel-body">
                 <div class="checkbox">
                     ${h.checkbox('rhodecode_pr_merge_enabled' + suffix, 'True', **kwargs)}
                     <label for="rhodecode_pr_merge_enabled${suffix}">${_('Enable server-side merge for pull requests')}</label>
                 </div>
                 <div class="label">
                     <span class="help-block">${_('Note: when this feature is enabled, it only runs hooks defined in the rcextension package. Custom hooks added on the Admin -> Settings -> Hooks page will not be run when pull requests are automatically merged from the web interface.')}</span>
                 </div>
                 <div class="checkbox">
                     ${h.checkbox('rhodecode_use_outdated_comments' + suffix, 'True', **kwargs)}
                     <label for="rhodecode_use_outdated_comments${suffix}">${_('Invalidate and relocate inline comments during update')}</label>
                 </div>
                 <div class="label">
                     <span class="help-block">${_('During the update of a pull request, the position of inline comments will be updated and outdated inline comments will be hidden.')}</span>
                 </div>
             </div>
         </div>
     % endif
 
 </%def>

rhodecode/tests/functional/test_admin_settings.py

@@ -1,655 +1,611 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  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 mock
 import pytest
 
 import rhodecode
 from rhodecode.config.routing import ADMIN_PREFIX
 from rhodecode.lib.utils2 import md5
 from rhodecode.model.db import RhodeCodeUi
 from rhodecode.model.meta import Session
 from rhodecode.model.settings import SettingsModel, IssueTrackerSettingsModel
 from rhodecode.tests import url, assert_session_flash
 from rhodecode.tests.utils import AssertResponse
 
 
 UPDATE_DATA_QUALNAME = (
     'rhodecode.controllers.admin.settings.SettingsController.get_update_data')
 
 
 @pytest.mark.usefixtures('autologin_user', 'app')
 class TestAdminSettingsController:
 
     @pytest.mark.parametrize('urlname', [
         'admin_settings_vcs',
         'admin_settings_mapping',
         'admin_settings_global',
         'admin_settings_visual',
         'admin_settings_email',
         'admin_settings_hooks',
         'admin_settings_search',
         'admin_settings_system',
     ])
     def test_simple_get(self, urlname, app):
         app.get(url(urlname))
 
     def test_create_custom_hook(self, csrf_token):
         response = self.app.post(
             url('admin_settings_hooks'),
             params={
                 'new_hook_ui_key': 'test_hooks_1',
                 'new_hook_ui_value': 'cd /tmp',
                 'csrf_token': csrf_token})
 
         response = response.follow()
         response.mustcontain('test_hooks_1')
         response.mustcontain('cd /tmp')
 
     def test_create_custom_hook_delete(self, csrf_token):
         response = self.app.post(
             url('admin_settings_hooks'),
             params={
                 'new_hook_ui_key': 'test_hooks_2',
                 'new_hook_ui_value': 'cd /tmp2',
                 'csrf_token': csrf_token})
 
         response = response.follow()
         response.mustcontain('test_hooks_2')
         response.mustcontain('cd /tmp2')
 
         hook_id = SettingsModel().get_ui_by_key('test_hooks_2').ui_id
 
         # delete
         self.app.post(
             url('admin_settings_hooks'),
             params={'hook_id': hook_id, 'csrf_token': csrf_token})
         response = self.app.get(url('admin_settings_hooks'))
         response.mustcontain(no=['test_hooks_2'])
         response.mustcontain(no=['cd /tmp2'])
 
     def test_system_update_new_version(self):
         update_data = {
             'versions': [
                 {
                     'version': '100.3.1415926535',
                     'general': 'The latest version we are ever going to ship'
                 },
                 {
                     'version': '0.0.0',
                     'general': 'The first version we ever shipped'
                 }
             ]
         }
         with mock.patch(UPDATE_DATA_QUALNAME, return_value=update_data):
             response = self.app.get(url('admin_settings_system_update'))
             response.mustcontain('A <b>new version</b> is available')
 
     def test_system_update_nothing_new(self):
         update_data = {
             'versions': [
                 {
                     'version': '0.0.0',
                     'general': 'The first version we ever shipped'
                 }
             ]
         }
         with mock.patch(UPDATE_DATA_QUALNAME, return_value=update_data):
             response = self.app.get(url('admin_settings_system_update'))
             response.mustcontain(
                 'You already have the <b>latest</b> stable version.')
 
     def test_system_update_bad_response(self):
         with mock.patch(UPDATE_DATA_QUALNAME, side_effect=ValueError('foo')):
             response = self.app.get(url('admin_settings_system_update'))
             response.mustcontain(
                 'Bad data sent from update server')
 
 
 @pytest.mark.usefixtures('autologin_user', 'app')
 class TestAdminSettingsGlobal:
 
     def test_pre_post_code_code_active(self, csrf_token):
         pre_code = 'rc-pre-code-187652122'
         post_code = 'rc-postcode-98165231'
 
         response = self.post_and_verify_settings({
             'rhodecode_pre_code': pre_code,
             'rhodecode_post_code': post_code,
             'csrf_token': csrf_token,
         })
 
         response = response.follow()
         response.mustcontain(pre_code, post_code)
 
     def test_pre_post_code_code_inactive(self, csrf_token):
         pre_code = 'rc-pre-code-187652122'
         post_code = 'rc-postcode-98165231'
         response = self.post_and_verify_settings({
             'rhodecode_pre_code': '',
             'rhodecode_post_code': '',
             'csrf_token': csrf_token,
         })
 
         response = response.follow()
         response.mustcontain(no=[pre_code, post_code])
 
     def test_captcha_activate(self, csrf_token):
         self.post_and_verify_settings({
             'rhodecode_captcha_private_key': '1234567890',
             'rhodecode_captcha_public_key': '1234567890',
             'csrf_token': csrf_token,
         })
 
         response = self.app.get(ADMIN_PREFIX + '/register')
         response.mustcontain('captcha')
 
     def test_captcha_deactivate(self, csrf_token):
         self.post_and_verify_settings({
             'rhodecode_captcha_private_key': '',
             'rhodecode_captcha_public_key': '1234567890',
             'csrf_token': csrf_token,
         })
 
         response = self.app.get(ADMIN_PREFIX + '/register')
         response.mustcontain(no=['captcha'])
 
     def test_title_change(self, csrf_token):
         old_title = 'RhodeCode'
         new_title = old_title + '_changed'
 
         for new_title in ['Changed', 'Żółwik', old_title]:
             response = self.post_and_verify_settings({
                 'rhodecode_title': new_title,
                 'csrf_token': csrf_token,
             })
 
             response = response.follow()
             response.mustcontain(
                 """<div class="branding">- %s</div>""" % new_title)
 
     def post_and_verify_settings(self, settings):
         old_title = 'RhodeCode'
         old_realm = 'RhodeCode authentication'
         params = {
             'rhodecode_title': old_title,
             'rhodecode_realm': old_realm,
             'rhodecode_pre_code': '',
             'rhodecode_post_code': '',
             'rhodecode_captcha_private_key': '',
             'rhodecode_captcha_public_key': '',
         }
         params.update(settings)
         response = self.app.post(url('admin_settings_global'), params=params)
 
         assert_session_flash(response, 'Updated application settings')
         app_settings = SettingsModel().get_all_settings()
         del settings['csrf_token']
         for key, value in settings.iteritems():
             assert app_settings[key] == value.decode('utf-8')
 
         return response
 
 
 @pytest.mark.usefixtures('autologin_user', 'app')
 class TestAdminSettingsVcs:
 
     def test_contains_svn_default_patterns(self, app):
         response = app.get(url('admin_settings_vcs'))
         expected_patterns = [
             '/trunk',
             '/branches/*',
             '/tags/*',
         ]
         for pattern in expected_patterns:
             response.mustcontain(pattern)
 
     def test_add_new_svn_branch_and_tag_pattern(
             self, app, backend_svn, form_defaults, disable_sql_cache,
             csrf_token):
         form_defaults.update({
             'new_svn_branch': '/exp/branches/*',
             'new_svn_tag': '/important_tags/*',
             'csrf_token': csrf_token,
         })
 
         response = app.post(
             url('admin_settings_vcs'), params=form_defaults, status=302)
         response = response.follow()
 
         # Expect to find the new values on the page
         response.mustcontain('/exp/branches/*')
         response.mustcontain('/important_tags/*')
 
         # Expect that those patterns are used to match branches and tags now
         repo = backend_svn['svn-simple-layout'].scm_instance()
         assert 'exp/branches/exp-sphinx-docs' in repo.branches
         assert 'important_tags/v0.5' in repo.tags
 
     def test_add_same_svn_value_twice_shows_an_error_message(
             self, app, form_defaults, csrf_token, settings_util):
         settings_util.create_rhodecode_ui('vcs_svn_branch', '/test')
         settings_util.create_rhodecode_ui('vcs_svn_tag', '/test')
 
         response = app.post(
             url('admin_settings_vcs'),
             params={
                 'paths_root_path': form_defaults['paths_root_path'],
                 'new_svn_branch': '/test',
                 'new_svn_tag': '/test',
                 'csrf_token': csrf_token,
             },
             status=200)
 
         response.mustcontain("Pattern already exists")
         response.mustcontain("Some form inputs contain invalid data.")
 
     @pytest.mark.parametrize('section', [
         'vcs_svn_branch',
         'vcs_svn_tag',
     ])
     def test_delete_svn_patterns(
             self, section, app, csrf_token, settings_util):
         setting = settings_util.create_rhodecode_ui(
             section, '/test_delete', cleanup=False)
 
         app.post(
             url('admin_settings_vcs'),
             params={
                 '_method': 'delete',
                 'delete_svn_pattern': setting.ui_id,
                 'csrf_token': csrf_token},
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest'})
 
     @pytest.mark.parametrize('section', [
         'vcs_svn_branch',
         'vcs_svn_tag',
     ])
     def test_delete_svn_patterns_raises_400_when_no_xhr(
             self, section, app, csrf_token, settings_util):
         setting = settings_util.create_rhodecode_ui(section, '/test_delete')
 
         app.post(
             url('admin_settings_vcs'),
             params={
                 '_method': 'delete',
                 'delete_svn_pattern': setting.ui_id,
                 'csrf_token': csrf_token},
             status=400)
 
     def test_extensions_hgsubversion(self, app, form_defaults, csrf_token):
         form_defaults.update({
             'csrf_token': csrf_token,
             'extensions_hgsubversion': 'True',
         })
         response = app.post(
             url('admin_settings_vcs'),
             params=form_defaults,
             status=302)
 
         response = response.follow()
         extensions_input = (
             '<input id="extensions_hgsubversion" '
             'name="extensions_hgsubversion" type="checkbox" '
             'value="True" checked="checked" />')
         response.mustcontain(extensions_input)
 
     def test_has_a_section_for_pull_request_settings(self, app):
         response = app.get(url('admin_settings_vcs'))
         response.mustcontain('Pull Request Settings')
 
     def test_has_an_input_for_invalidation_of_inline_comments(
             self, app):
         response = app.get(url('admin_settings_vcs'))
         assert_response = AssertResponse(response)
         assert_response.one_element_exists(
             '[name=rhodecode_use_outdated_comments]')
 
     @pytest.mark.parametrize('new_value', [True, False])
     def test_allows_to_change_invalidation_of_inline_comments(
             self, app, form_defaults, csrf_token, new_value):
         setting_key = 'use_outdated_comments'
         setting = SettingsModel().create_or_update_setting(
             setting_key, not new_value, 'bool')
         Session().add(setting)
         Session().commit()
 
         form_defaults.update({
             'csrf_token': csrf_token,
             'rhodecode_use_outdated_comments': str(new_value),
         })
         response = app.post(
             url('admin_settings_vcs'),
             params=form_defaults,
             status=302)
         response = response.follow()
         setting = SettingsModel().get_setting_by_name(setting_key)
         assert setting.app_settings_value is new_value
 
     def test_has_a_section_for_labs_settings_if_enabled(self, app):
         with mock.patch.dict(
                 rhodecode.CONFIG, {'labs_settings_active': 'true'}):
             response = self.app.get(url('admin_settings_vcs'))
-        response.mustcontain('Labs settings:')
+        response.mustcontain('Labs Settings')
 
     def test_has_not_a_section_for_labs_settings_if_disables(self, app):
         with mock.patch.dict(
                 rhodecode.CONFIG, {'labs_settings_active': 'false'}):
             response = self.app.get(url('admin_settings_vcs'))
-        response.mustcontain(no='Labs settings:')
+        response.mustcontain(no='Labs Settings')
 
     @pytest.mark.parametrize('new_value', [True, False])
     def test_allows_to_change_hg_rebase_merge_strategy(
             self, app, form_defaults, csrf_token, new_value):
         setting_key = 'hg_use_rebase_for_merging'
 
         form_defaults.update({
             'csrf_token': csrf_token,
             'rhodecode_' + setting_key: str(new_value),
         })
 
         with mock.patch.dict(
                 rhodecode.CONFIG, {'labs_settings_active': 'true'}):
             app.post(
                 url('admin_settings_vcs'),
                 params=form_defaults,
                 status=302)
 
         setting = SettingsModel().get_setting_by_name(setting_key)
         assert setting.app_settings_value is new_value
 
     @pytest.fixture
     def disable_sql_cache(self, request):
         patcher = mock.patch(
             'rhodecode.lib.caching_query.FromCache.process_query')
         request.addfinalizer(patcher.stop)
         patcher.start()
 
     @pytest.fixture
     def form_defaults(self):
         from rhodecode.controllers.admin.settings import SettingsController
         controller = SettingsController()
         return controller._form_defaults()
 
     # TODO: johbo: What we really want is to checkpoint before a test run and
     # reset the session afterwards.
     @pytest.fixture(scope='class', autouse=True)
     def cleanup_settings(self, request, pylonsapp):
         ui_id = RhodeCodeUi.ui_id
         original_ids = list(
             r.ui_id for r in RhodeCodeUi.query().values(ui_id))
 
         @request.addfinalizer
         def cleanup():
             RhodeCodeUi.query().filter(
                 ui_id.notin_(original_ids)).delete(False)
 
 
 @pytest.mark.usefixtures('autologin_user', 'app')
 class TestLabsSettings(object):
     def test_get_settings_page_disabled(self):
         with mock.patch.dict(rhodecode.CONFIG,
                              {'labs_settings_active': 'false'}):
             response = self.app.get(url('admin_settings_labs'), status=302)
 
         assert response.location.endswith(url('admin_settings'))
 
     def test_get_settings_page_enabled(self):
         from rhodecode.controllers.admin import settings
         lab_settings = [
             settings.LabSetting(
                 key='rhodecode_bool',
                 type='bool',
                 group='bool group',
                 label='bool label',
                 help='bool help'
             ),
             settings.LabSetting(
                 key='rhodecode_text',
                 type='unicode',
                 group='text group',
                 label='text label',
                 help='text help'
             ),
         ]
         with mock.patch.dict(rhodecode.CONFIG,
                              {'labs_settings_active': 'true'}):
             with mock.patch.object(settings, '_LAB_SETTINGS', lab_settings):
                 response = self.app.get(url('admin_settings_labs'))
 
         assert '<label>bool group:</label>' in response
         assert '<label for="rhodecode_bool">bool label</label>' in response
         assert '<p class="help-block">bool help</p>' in response
         assert 'name="rhodecode_bool" type="checkbox"' in response
 
         assert '<label>text group:</label>' in response
         assert '<label for="rhodecode_text">text label</label>' in response
         assert '<p class="help-block">text help</p>' in response
         assert 'name="rhodecode_text" size="60" type="text"' in response
 
-    @pytest.mark.parametrize('setting_name', [
-        'proxy_subversion_http_requests',
-    ])
-    def test_update_boolean_settings(self, csrf_token, setting_name):
-        self.app.post(
-            url('admin_settings_labs'),
-            params={
-                'rhodecode_{}'.format(setting_name): 'true',
-                'csrf_token': csrf_token,
-            })
-        setting = SettingsModel().get_setting_by_name(setting_name)
-        assert setting.app_settings_value
-
-        self.app.post(
-            url('admin_settings_labs'),
-            params={
-                'rhodecode_{}'.format(setting_name): 'false',
-                'csrf_token': csrf_token,
-            })
-        setting = SettingsModel().get_setting_by_name(setting_name)
-        assert not setting.app_settings_value
-
-    @pytest.mark.parametrize('setting_name', [
-        'subversion_http_server_url',
-    ])
-    def test_update_string_settings(self, csrf_token, setting_name):
-        self.app.post(
-            url('admin_settings_labs'),
-            params={
-                'rhodecode_{}'.format(setting_name): 'Test 1',
-                'csrf_token': csrf_token,
-            })
-        setting = SettingsModel().get_setting_by_name(setting_name)
-        assert setting.app_settings_value == 'Test 1'
-
-        self.app.post(
-            url('admin_settings_labs'),
-            params={
-                'rhodecode_{}'.format(setting_name): ' Test 2 ',
-                'csrf_token': csrf_token,
-            })
-        setting = SettingsModel().get_setting_by_name(setting_name)
-        assert setting.app_settings_value == 'Test 2'
-
 
 @pytest.mark.usefixtures('app')
 class TestOpenSourceLicenses(object):
 
     def _get_url(self):
         return ADMIN_PREFIX + '/settings/open_source'
 
     def test_records_are_displayed(self, autologin_user):
         sample_licenses = {
             "python2.7-pytest-2.7.1": {
                 "UNKNOWN": None
             },
             "python2.7-Markdown-2.6.2": {
                 "BSD-3-Clause": "http://spdx.org/licenses/BSD-3-Clause"
             }
         }
         read_licenses_patch = mock.patch(
             'rhodecode.admin.views.read_opensource_licenses',
             return_value=sample_licenses)
         with read_licenses_patch:
             response = self.app.get(self._get_url(), status=200)
 
         assert_response = AssertResponse(response)
         assert_response.element_contains(
             '.panel-heading', 'Licenses of Third Party Packages')
         for name in sample_licenses:
             response.mustcontain(name)
             for license in sample_licenses[name]:
                 assert_response.element_contains('.panel-body', license)
 
     def test_records_can_be_read(self, autologin_user):
         response = self.app.get(self._get_url(), status=200)
         assert_response = AssertResponse(response)
         assert_response.element_contains(
             '.panel-heading', 'Licenses of Third Party Packages')
 
     def test_forbidden_when_normal_user(self, autologin_regular_user):
         self.app.get(self._get_url(), status=403)
 
 
 @pytest.mark.usefixtures("app")
 class TestAdminSettingsIssueTracker:
     RC_PREFIX = 'rhodecode_'
     SHORT_PATTERN_KEY = 'issuetracker_pat_'
     PATTERN_KEY = RC_PREFIX + SHORT_PATTERN_KEY
 
     def test_issuetracker_index(self, autologin_user):
         response = self.app.get(url('admin_settings_issuetracker'))
         assert response.status_code == 200
 
     def test_add_issuetracker_pattern(
             self, request, autologin_user, csrf_token):
         pattern = 'issuetracker_pat'
         another_pattern = pattern+'1'
         post_url = url('admin_settings_issuetracker_save')
         post_data = {
             'new_pattern_pattern_0': pattern,
             'new_pattern_url_0': 'url',
             'new_pattern_prefix_0': 'prefix',
             'new_pattern_description_0': 'description',
             'new_pattern_pattern_1': another_pattern,
             'new_pattern_url_1': 'url1',
             'new_pattern_prefix_1': 'prefix1',
             'new_pattern_description_1': 'description1',
             'csrf_token': csrf_token
         }
         self.app.post(post_url, post_data, status=302)
         settings = SettingsModel().get_all_settings()
         self.uid = md5(pattern)
         assert settings[self.PATTERN_KEY+self.uid] == pattern
         self.another_uid = md5(another_pattern)
         assert settings[self.PATTERN_KEY+self.another_uid] == another_pattern
 
         @request.addfinalizer
         def cleanup():
             defaults = SettingsModel().get_all_settings()
 
             entries = [name for name in defaults if (
                 (self.uid in name) or (self.another_uid) in name)]
             start = len(self.RC_PREFIX)
             for del_key in entries:
                 # TODO: anderson: get_by_name needs name without prefix
                 entry = SettingsModel().get_setting_by_name(del_key[start:])
                 Session().delete(entry)
 
             Session().commit()
 
     def test_edit_issuetracker_pattern(
             self, autologin_user, backend, csrf_token, request):
         old_pattern = 'issuetracker_pat'
         old_uid = md5(old_pattern)
         pattern = 'issuetracker_pat_new'
         self.new_uid = md5(pattern)
 
         SettingsModel().create_or_update_setting(
             self.SHORT_PATTERN_KEY+old_uid, old_pattern, 'unicode')
 
         post_url = url('admin_settings_issuetracker_save')
         post_data = {
             'new_pattern_pattern_0': pattern,
             'new_pattern_url_0': 'url',
             'new_pattern_prefix_0': 'prefix',
             'new_pattern_description_0': 'description',
             'uid': old_uid,
             'csrf_token': csrf_token
         }
         self.app.post(post_url, post_data, status=302)
         settings = SettingsModel().get_all_settings()
         assert settings[self.PATTERN_KEY+self.new_uid] == pattern
         assert self.PATTERN_KEY+old_uid not in settings
 
         @request.addfinalizer
         def cleanup():
             IssueTrackerSettingsModel().delete_entries(self.new_uid)
 
     def test_replace_issuetracker_pattern_description(
             self, autologin_user, csrf_token, request, settings_util):
         prefix = 'issuetracker'
         pattern = 'issuetracker_pat'
         self.uid = md5(pattern)
         pattern_key = '_'.join([prefix, 'pat', self.uid])
         rc_pattern_key = '_'.join(['rhodecode', pattern_key])
         desc_key = '_'.join([prefix, 'desc', self.uid])
         rc_desc_key = '_'.join(['rhodecode', desc_key])
         new_description = 'new_description'
 
         settings_util.create_rhodecode_setting(
             pattern_key, pattern, 'unicode', cleanup=False)
         settings_util.create_rhodecode_setting(
             desc_key, 'old description', 'unicode', cleanup=False)
 
         post_url = url('admin_settings_issuetracker_save')
         post_data = {
             'new_pattern_pattern_0': pattern,
             'new_pattern_url_0': 'url',
             'new_pattern_prefix_0': 'prefix',
             'new_pattern_description_0': new_description,
             'uid': self.uid,
             'csrf_token': csrf_token
         }
         self.app.post(post_url, post_data, status=302)
         settings = SettingsModel().get_all_settings()
         assert settings[rc_pattern_key] == pattern
         assert settings[rc_desc_key] == new_description
 
         @request.addfinalizer
         def cleanup():
             IssueTrackerSettingsModel().delete_entries(self.uid)
 
     def test_delete_issuetracker_pattern(
             self, autologin_user, backend, csrf_token, settings_util):
         pattern = 'issuetracker_pat'
         uid = md5(pattern)
         settings_util.create_rhodecode_setting(
             self.SHORT_PATTERN_KEY+uid, pattern, 'unicode', cleanup=False)
 
         post_url = url('admin_issuetracker_delete')
         post_data = {
             '_method': 'delete',
             'uid': uid,
             'csrf_token': csrf_token
         }
         self.app.post(post_url, post_data, status=302)
         settings = SettingsModel().get_all_settings()
         assert 'rhodecode_%s%s' % (self.SHORT_PATTERN_KEY, uid) not in settings

rhodecode/tests/lib/middleware/test_simplesvn.py

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

rhodecode/tests/lib/middleware/test_vcs.py

@@ -1,136 +1,140 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 from mock import patch, Mock
 
 import rhodecode
 from rhodecode.lib.middleware import vcs
+from rhodecode.lib.middleware.simplesvn import (
+    SimpleSvn, DisabledSimpleSvnApp, SimpleSvnApp)
+from rhodecode.tests import SVN_REPO
 
+svn_repo_path = '/'+ SVN_REPO
 
 def test_is_hg():
     environ = {
-        'PATH_INFO': '/rhodecode-dev',
+        'PATH_INFO': svn_repo_path,
         'QUERY_STRING': 'cmd=changegroup',
         'HTTP_ACCEPT': 'application/mercurial'
     }
     assert vcs.is_hg(environ)
 
 
 def test_is_hg_no_cmd():
     environ = {
-        'PATH_INFO': '/rhodecode-dev',
+        'PATH_INFO': svn_repo_path,
         'QUERY_STRING': '',
         'HTTP_ACCEPT': 'application/mercurial'
     }
     assert not vcs.is_hg(environ)
 
 
 def test_is_hg_empty_cmd():
     environ = {
-        'PATH_INFO': '/rhodecode-dev',
+        'PATH_INFO': svn_repo_path,
         'QUERY_STRING': 'cmd=',
         'HTTP_ACCEPT': 'application/mercurial'
     }
     assert not vcs.is_hg(environ)
 
 
 def test_is_svn_returns_true_if_subversion_is_in_a_dav_header():
     environ = {
-        'PATH_INFO': '/rhodecode-dev',
+        'PATH_INFO': svn_repo_path,
         'HTTP_DAV': 'http://subversion.tigris.org/xmlns/dav/svn/log-revprops'
     }
     assert vcs.is_svn(environ) is True
 
 
 def test_is_svn_returns_false_if_subversion_is_not_in_a_dav_header():
     environ = {
-        'PATH_INFO': '/rhodecode-dev',
+        'PATH_INFO': svn_repo_path,
         'HTTP_DAV': 'http://stuff.tigris.org/xmlns/dav/svn/log-revprops'
     }
     assert vcs.is_svn(environ) is False
 
 
 def test_is_svn_returns_false_if_no_dav_header():
     environ = {
-        'PATH_INFO': '/rhodecode-dev',
+        'PATH_INFO': svn_repo_path,
     }
     assert vcs.is_svn(environ) is False
 
 
 def test_is_svn_returns_true_if_magic_path_segment():
     environ = {
         'PATH_INFO': '/stub-repository/!svn/rev/4',
     }
     assert vcs.is_svn(environ)
 
 
 def test_is_svn_allows_to_configure_the_magic_path(monkeypatch):
     """
     This is intended as a fallback in case someone has configured his
     Subversion server with a different magic path segment.
     """
     monkeypatch.setitem(
         rhodecode.CONFIG, 'rhodecode_subversion_magic_path', '/!my-magic')
     environ = {
         'PATH_INFO': '/stub-repository/!my-magic/rev/4',
     }
     assert vcs.is_svn(environ)
 
 
 class TestVCSMiddleware(object):
-    def test_get_handler_app_retuns_svn_app_when_proxy_enabled(self):
+    def test_get_handler_app_retuns_svn_app_when_proxy_enabled(self, app):
         environ = {
-            'PATH_INFO': 'rhodecode-dev',
+            'PATH_INFO': SVN_REPO,
             'HTTP_DAV': 'http://subversion.tigris.org/xmlns/dav/svn/log'
         }
-        app = Mock()
-        config = Mock()
+        application = Mock()
+        config = {'appenlight': False}
         registry = Mock()
         middleware = vcs.VCSMiddleware(
-            app, config=config, appenlight_client=None, registry=registry)
-        snv_patch = patch('rhodecode.lib.middleware.vcs.SimpleSvn')
-        settings_patch = patch.dict(
-            rhodecode.CONFIG,
-            {'rhodecode_proxy_subversion_http_requests': True})
-        with snv_patch as svn_mock, settings_patch:
-            svn_mock.return_value = None
-            middleware._get_handler_app(environ)
+            application, config=config,
+            appenlight_client=None, registry=registry)
+        middleware.use_gzip = False
 
-        svn_mock.assert_called_once_with(app, config, registry)
+        with patch.object(SimpleSvn, '_is_svn_enabled') as mock_method:
+            mock_method.return_value = True
+            application = middleware._get_handler_app(environ)
+            assert isinstance(application, SimpleSvn)
+            assert isinstance(application._create_wsgi_app(
+                Mock(), Mock(), Mock()), SimpleSvnApp)
 
-    def test_get_handler_app_retuns_no_svn_app_when_proxy_disabled(self):
+    def test_get_handler_app_retuns_dummy_svn_app_when_proxy_disabled(self, app):
         environ = {
-            'PATH_INFO': 'rhodecode-dev',
+            'PATH_INFO': SVN_REPO,
             'HTTP_DAV': 'http://subversion.tigris.org/xmlns/dav/svn/log'
         }
-        app = Mock()
-        config = Mock()
+        application = Mock()
+        config = {'appenlight': False}
         registry = Mock()
         middleware = vcs.VCSMiddleware(
-            app, config=config, appenlight_client=None, registry=registry)
-        snv_patch = patch('rhodecode.lib.middleware.vcs.SimpleSvn')
-        settings_patch = patch.dict(
-            rhodecode.CONFIG,
-            {'rhodecode_proxy_subversion_http_requests': False})
-        with snv_patch as svn_mock, settings_patch:
-            app = middleware._get_handler_app(environ)
+            application, config=config,
+            appenlight_client=None, registry=registry)
+        middleware.use_gzip = False
 
-        assert svn_mock.call_count == 0
-        assert app is None
+        with patch.object(SimpleSvn, '_is_svn_enabled') as mock_method:
+            mock_method.return_value = False
+            application = middleware._get_handler_app(environ)
+            assert isinstance(application, SimpleSvn)
+            assert isinstance(application._create_wsgi_app(
+                Mock(), Mock(), Mock()), DisabledSimpleSvnApp)

rhodecode/tests/models/settings/test_vcs_settings.py

@@ -1,1038 +1,1029 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  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 mock
 import pytest
 
 from rhodecode.lib.utils2 import str2bool
 from rhodecode.model.meta import Session
 from rhodecode.model.settings import VcsSettingsModel, UiSetting
 
 
 HOOKS_FORM_DATA = {
     'hooks_changegroup_repo_size': True,
     'hooks_changegroup_push_logger': True,
     'hooks_outgoing_pull_logger': True
 }
 
 SVN_FORM_DATA = {
     'new_svn_branch': 'test-branch',
     'new_svn_tag': 'test-tag'
 }
 
 GENERAL_FORM_DATA = {
     'rhodecode_pr_merge_enabled': True,
     'rhodecode_use_outdated_comments': True,
     'rhodecode_hg_use_rebase_for_merging': True,
 }
 
 
 class TestInheritGlobalSettingsProperty(object):
     def test_get_raises_exception_when_repository_not_specified(self):
         model = VcsSettingsModel()
         with pytest.raises(Exception) as exc_info:
             model.inherit_global_settings
         assert exc_info.value.message == 'Repository is not specified'
 
     def test_true_is_returned_when_value_is_not_found(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         assert model.inherit_global_settings is True
 
     def test_value_is_returned(self, repo_stub, settings_util):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         settings_util.create_repo_rhodecode_setting(
             repo_stub, VcsSettingsModel.INHERIT_SETTINGS, False, 'bool')
         assert model.inherit_global_settings is False
 
     def test_value_is_set(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         model.inherit_global_settings = False
         setting = model.repo_settings.get_setting_by_name(
             VcsSettingsModel.INHERIT_SETTINGS)
         try:
             assert setting.app_settings_type == 'bool'
             assert setting.app_settings_value is False
         finally:
             Session().delete(setting)
             Session().commit()
 
     def test_set_raises_exception_when_repository_not_specified(self):
         model = VcsSettingsModel()
         with pytest.raises(Exception) as exc_info:
             model.inherit_global_settings = False
         assert exc_info.value.message == 'Repository is not specified'
 
 
 class TestVcsSettingsModel(object):
     def test_global_svn_branch_patterns(self):
         model = VcsSettingsModel()
         expected_result = {'test': 'test'}
         with mock.patch.object(model, 'global_settings') as settings_mock:
             get_settings = settings_mock.get_ui_by_section
             get_settings.return_value = expected_result
             settings_mock.return_value = expected_result
             result = model.get_global_svn_branch_patterns()
 
         get_settings.assert_called_once_with(model.SVN_BRANCH_SECTION)
         assert expected_result == result
 
     def test_repo_svn_branch_patterns(self):
         model = VcsSettingsModel()
         expected_result = {'test': 'test'}
         with mock.patch.object(model, 'repo_settings') as settings_mock:
             get_settings = settings_mock.get_ui_by_section
             get_settings.return_value = expected_result
             settings_mock.return_value = expected_result
             result = model.get_repo_svn_branch_patterns()
 
         get_settings.assert_called_once_with(model.SVN_BRANCH_SECTION)
         assert expected_result == result
 
     def test_repo_svn_branch_patterns_raises_exception_when_repo_is_not_set(
             self):
         model = VcsSettingsModel()
         with pytest.raises(Exception) as exc_info:
             model.get_repo_svn_branch_patterns()
         assert exc_info.value.message == 'Repository is not specified'
 
     def test_global_svn_tag_patterns(self):
         model = VcsSettingsModel()
         expected_result = {'test': 'test'}
         with mock.patch.object(model, 'global_settings') as settings_mock:
             get_settings = settings_mock.get_ui_by_section
             get_settings.return_value = expected_result
             settings_mock.return_value = expected_result
             result = model.get_global_svn_tag_patterns()
 
         get_settings.assert_called_once_with(model.SVN_TAG_SECTION)
         assert expected_result == result
 
     def test_repo_svn_tag_patterns(self):
         model = VcsSettingsModel()
         expected_result = {'test': 'test'}
         with mock.patch.object(model, 'repo_settings') as settings_mock:
             get_settings = settings_mock.get_ui_by_section
             get_settings.return_value = expected_result
             settings_mock.return_value = expected_result
             result = model.get_repo_svn_tag_patterns()
 
         get_settings.assert_called_once_with(model.SVN_TAG_SECTION)
         assert expected_result == result
 
     def test_repo_svn_tag_patterns_raises_exception_when_repo_is_not_set(self):
         model = VcsSettingsModel()
         with pytest.raises(Exception) as exc_info:
             model.get_repo_svn_tag_patterns()
         assert exc_info.value.message == 'Repository is not specified'
 
     def test_get_global_settings(self):
         expected_result = {'test': 'test'}
         model = VcsSettingsModel()
         with mock.patch.object(model, '_collect_all_settings') as collect_mock:
             collect_mock.return_value = expected_result
             result = model.get_global_settings()
 
         collect_mock.assert_called_once_with(global_=True)
         assert result == expected_result
 
     def test_get_repo_settings(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         expected_result = {'test': 'test'}
         with mock.patch.object(model, '_collect_all_settings') as collect_mock:
             collect_mock.return_value = expected_result
             result = model.get_repo_settings()
 
         collect_mock.assert_called_once_with(global_=False)
         assert result == expected_result
 
     @pytest.mark.parametrize('settings, global_', [
         ('global_settings', True),
         ('repo_settings', False)
     ])
     def test_collect_all_settings(self, settings, global_):
         model = VcsSettingsModel()
         result_mock = self._mock_result()
 
         settings_patch = mock.patch.object(model, settings)
         with settings_patch as settings_mock:
             settings_mock.get_ui_by_section_and_key.return_value = result_mock
             settings_mock.get_setting_by_name.return_value = result_mock
             result = model._collect_all_settings(global_=global_)
 
         ui_settings = model.HG_SETTINGS + model.HOOKS_SETTINGS
         self._assert_get_settings_calls(
             settings_mock, ui_settings, model.GENERAL_SETTINGS)
         self._assert_collect_all_settings_result(
             ui_settings, model.GENERAL_SETTINGS, result)
 
     @pytest.mark.parametrize('settings, global_', [
         ('global_settings', True),
         ('repo_settings', False)
     ])
     def test_collect_all_settings_without_empty_value(self, settings, global_):
         model = VcsSettingsModel()
 
         settings_patch = mock.patch.object(model, settings)
         with settings_patch as settings_mock:
             settings_mock.get_ui_by_section_and_key.return_value = None
             settings_mock.get_setting_by_name.return_value = None
             result = model._collect_all_settings(global_=global_)
 
         assert result == {}
 
     def _mock_result(self):
         result_mock = mock.Mock()
         result_mock.ui_value = 'ui_value'
         result_mock.ui_active = True
         result_mock.app_settings_value = 'setting_value'
         return result_mock
 
     def _assert_get_settings_calls(
             self, settings_mock, ui_settings, general_settings):
         assert (
             settings_mock.get_ui_by_section_and_key.call_count ==
             len(ui_settings))
         assert (
             settings_mock.get_setting_by_name.call_count ==
             len(general_settings))
 
         for section, key in ui_settings:
             expected_call = mock.call(section, key)
             assert (
                 expected_call in
                 settings_mock.get_ui_by_section_and_key.call_args_list)
 
         for name in general_settings:
             expected_call = mock.call(name)
             assert (
                 expected_call in
                 settings_mock.get_setting_by_name.call_args_list)
 
     def _assert_collect_all_settings_result(
             self, ui_settings, general_settings, result):
         expected_result = {}
         for section, key in ui_settings:
             key = '{}_{}'.format(section, key.replace('.', '_'))
             value = True if section in ('extensions', 'hooks') else 'ui_value'
             expected_result[key] = value
 
         for name in general_settings:
             key = 'rhodecode_' + name
             expected_result[key] = 'setting_value'
 
         assert expected_result == result
 
 
 class TestCreateOrUpdateRepoHookSettings(object):
     def test_create_when_no_repo_object_found(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
 
         self._create_settings(model, HOOKS_FORM_DATA)
 
         cleanup = []
         try:
             for section, key in model.HOOKS_SETTINGS:
                 ui = model.repo_settings.get_ui_by_section_and_key(
                     section, key)
                 assert ui.ui_active is True
                 cleanup.append(ui)
         finally:
             for ui in cleanup:
                 Session().delete(ui)
             Session().commit()
 
     def test_create_raises_exception_when_data_incomplete(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
 
         deleted_key = 'hooks_changegroup_repo_size'
         data = HOOKS_FORM_DATA.copy()
         data.pop(deleted_key)
 
         with pytest.raises(ValueError) as exc_info:
             model.create_or_update_repo_hook_settings(data)
         assert (
             exc_info.value.message ==
             'The given data does not contain {} key'.format(deleted_key))
 
     def test_update_when_repo_object_found(self, repo_stub, settings_util):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         for section, key in model.HOOKS_SETTINGS:
             settings_util.create_repo_rhodecode_ui(
                 repo_stub, section, None, key=key, active=False)
         model.create_or_update_repo_hook_settings(HOOKS_FORM_DATA)
         for section, key in model.HOOKS_SETTINGS:
             ui = model.repo_settings.get_ui_by_section_and_key(section, key)
             assert ui.ui_active is True
 
     def _create_settings(self, model, data):
         global_patch = mock.patch.object(model, 'global_settings')
         global_setting = mock.Mock()
         global_setting.ui_value = 'Test value'
         with global_patch as global_mock:
             global_mock.get_ui_by_section_and_key.return_value = global_setting
             model.create_or_update_repo_hook_settings(HOOKS_FORM_DATA)
 
 
 class TestUpdateGlobalHookSettings(object):
     def test_update_raises_exception_when_data_incomplete(self):
         model = VcsSettingsModel()
 
         deleted_key = 'hooks_changegroup_repo_size'
         data = HOOKS_FORM_DATA.copy()
         data.pop(deleted_key)
 
         with pytest.raises(ValueError) as exc_info:
             model.update_global_hook_settings(data)
         assert (
             exc_info.value.message ==
             'The given data does not contain {} key'.format(deleted_key))
 
     def test_update_global_hook_settings(self, settings_util):
         model = VcsSettingsModel()
         setting_mock = mock.MagicMock()
         setting_mock.ui_active = False
         get_settings_patcher = mock.patch.object(
             model.global_settings, 'get_ui_by_section_and_key',
             return_value=setting_mock)
         session_patcher = mock.patch('rhodecode.model.settings.Session')
         with get_settings_patcher as get_settings_mock, session_patcher:
             model.update_global_hook_settings(HOOKS_FORM_DATA)
         assert setting_mock.ui_active is True
         assert get_settings_mock.call_count == 3
 
 
 class TestCreateOrUpdateRepoGeneralSettings(object):
     def test_calls_create_or_update_general_settings(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         create_patch = mock.patch.object(
             model, '_create_or_update_general_settings')
         with create_patch as create_mock:
             model.create_or_update_repo_pr_settings(GENERAL_FORM_DATA)
         create_mock.assert_called_once_with(
             model.repo_settings, GENERAL_FORM_DATA)
 
     def test_raises_exception_when_repository_is_not_specified(self):
         model = VcsSettingsModel()
         with pytest.raises(Exception) as exc_info:
             model.create_or_update_repo_pr_settings(GENERAL_FORM_DATA)
         assert exc_info.value.message == 'Repository is not specified'
 
 
 class TestCreateOrUpdatGlobalGeneralSettings(object):
     def test_calls_create_or_update_general_settings(self):
         model = VcsSettingsModel()
         create_patch = mock.patch.object(
             model, '_create_or_update_general_settings')
         with create_patch as create_mock:
             model.create_or_update_global_pr_settings(GENERAL_FORM_DATA)
         create_mock.assert_called_once_with(
             model.global_settings, GENERAL_FORM_DATA)
 
 
 class TestCreateOrUpdateGeneralSettings(object):
     def test_create_when_no_repo_settings_found(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         model._create_or_update_general_settings(
             model.repo_settings, GENERAL_FORM_DATA)
 
         cleanup = []
         try:
             for name in model.GENERAL_SETTINGS:
                 setting = model.repo_settings.get_setting_by_name(name)
                 assert setting.app_settings_value is True
                 cleanup.append(setting)
         finally:
             for setting in cleanup:
                 Session().delete(setting)
             Session().commit()
 
     def test_create_raises_exception_when_data_incomplete(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
 
         deleted_key = 'rhodecode_pr_merge_enabled'
         data = GENERAL_FORM_DATA.copy()
         data.pop(deleted_key)
 
         with pytest.raises(ValueError) as exc_info:
             model._create_or_update_general_settings(model.repo_settings, data)
         assert (
             exc_info.value.message ==
             'The given data does not contain {} key'.format(deleted_key))
 
     def test_update_when_repo_setting_found(self, repo_stub, settings_util):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         for name in model.GENERAL_SETTINGS:
             settings_util.create_repo_rhodecode_setting(
                 repo_stub, name, False, 'bool')
 
         model._create_or_update_general_settings(
             model.repo_settings, GENERAL_FORM_DATA)
 
         for name in model.GENERAL_SETTINGS:
             setting = model.repo_settings.get_setting_by_name(name)
             assert setting.app_settings_value is True
 
 
 class TestCreateRepoSvnSettings(object):
     def test_calls_create_svn_settings(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         with mock.patch.object(model, '_create_svn_settings') as create_mock:
             model.create_repo_svn_settings(SVN_FORM_DATA)
         create_mock.assert_called_once_with(model.repo_settings, SVN_FORM_DATA)
 
     def test_raises_exception_when_repository_is_not_specified(self):
         model = VcsSettingsModel()
         with pytest.raises(Exception) as exc_info:
             model.create_repo_svn_settings(SVN_FORM_DATA)
         assert exc_info.value.message == 'Repository is not specified'
 
 
-class TestCreateGlobalSvnSettings(object):
-    def test_calls_create_svn_settings(self):
-        model = VcsSettingsModel()
-        with mock.patch.object(model, '_create_svn_settings') as create_mock:
-            model.create_global_svn_settings(SVN_FORM_DATA)
-        create_mock.assert_called_once_with(
-            model.global_settings, SVN_FORM_DATA)
-
-
 class TestCreateSvnSettings(object):
     def test_create(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         model._create_svn_settings(model.repo_settings, SVN_FORM_DATA)
         Session().commit()
 
         branch_ui = model.repo_settings.get_ui_by_section(
             model.SVN_BRANCH_SECTION)
         tag_ui = model.repo_settings.get_ui_by_section(
             model.SVN_TAG_SECTION)
 
         try:
             assert len(branch_ui) == 1
             assert len(tag_ui) == 1
         finally:
             Session().delete(branch_ui[0])
             Session().delete(tag_ui[0])
             Session().commit()
 
     def test_create_tag(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         data = SVN_FORM_DATA.copy()
         data.pop('new_svn_branch')
         model._create_svn_settings(model.repo_settings, data)
         Session().commit()
 
         branch_ui = model.repo_settings.get_ui_by_section(
             model.SVN_BRANCH_SECTION)
         tag_ui = model.repo_settings.get_ui_by_section(
             model.SVN_TAG_SECTION)
 
         try:
             assert len(branch_ui) == 0
             assert len(tag_ui) == 1
         finally:
             Session().delete(tag_ui[0])
             Session().commit()
 
     def test_create_nothing_when_no_svn_settings_specified(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         model._create_svn_settings(model.repo_settings, {})
         Session().commit()
 
         branch_ui = model.repo_settings.get_ui_by_section(
             model.SVN_BRANCH_SECTION)
         tag_ui = model.repo_settings.get_ui_by_section(
             model.SVN_TAG_SECTION)
 
         assert len(branch_ui) == 0
         assert len(tag_ui) == 0
 
     def test_create_nothing_when_empty_settings_specified(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         data = {
             'new_svn_branch': '',
             'new_svn_tag': ''
         }
         model._create_svn_settings(model.repo_settings, data)
         Session().commit()
 
         branch_ui = model.repo_settings.get_ui_by_section(
             model.SVN_BRANCH_SECTION)
         tag_ui = model.repo_settings.get_ui_by_section(
             model.SVN_TAG_SECTION)
 
         assert len(branch_ui) == 0
         assert len(tag_ui) == 0
 
 
 class TestCreateOrUpdateUi(object):
     def test_create(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         model._create_or_update_ui(
             model.repo_settings, 'test-section', 'test-key', active=False,
             value='False')
         Session().commit()
 
         created_ui = model.repo_settings.get_ui_by_section_and_key(
             'test-section', 'test-key')
 
         try:
             assert created_ui.ui_active is False
             assert str2bool(created_ui.ui_value) is False
         finally:
             Session().delete(created_ui)
             Session().commit()
 
     def test_update(self, repo_stub, settings_util):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
 
         largefiles, phases = model.HG_SETTINGS
         section = 'test-section'
         key = 'test-key'
         settings_util.create_repo_rhodecode_ui(
             repo_stub, section, 'True', key=key, active=True)
 
         model._create_or_update_ui(
             model.repo_settings, section, key, active=False, value='False')
         Session().commit()
 
         created_ui = model.repo_settings.get_ui_by_section_and_key(
             section, key)
         assert created_ui.ui_active is False
         assert str2bool(created_ui.ui_value) is False
 
 
 class TestCreateOrUpdateRepoHgSettings(object):
     FORM_DATA = {
         'extensions_largefiles': False,
         'phases_publish': False
     }
 
     def test_creates_repo_hg_settings_when_data_is_correct(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         with mock.patch.object(model, '_create_or_update_ui') as create_mock:
             model.create_or_update_repo_hg_settings(self.FORM_DATA)
         expected_calls = [
             mock.call(model.repo_settings, 'extensions', 'largefiles',
                       active=False, value=''),
             mock.call(model.repo_settings, 'phases', 'publish', value='False'),
         ]
         assert expected_calls == create_mock.call_args_list
 
     @pytest.mark.parametrize('field_to_remove', FORM_DATA.keys())
     def test_key_is_not_found(self, repo_stub, field_to_remove):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         data = self.FORM_DATA.copy()
         data.pop(field_to_remove)
         with pytest.raises(ValueError) as exc_info:
             model.create_or_update_repo_hg_settings(data)
         expected_message = 'The given data does not contain {} key'.format(
             field_to_remove)
         assert exc_info.value.message == expected_message
 
     def test_create_raises_exception_when_repository_not_specified(self):
         model = VcsSettingsModel()
         with pytest.raises(Exception) as exc_info:
             model.create_or_update_repo_hg_settings(self.FORM_DATA)
         assert exc_info.value.message == 'Repository is not specified'
 
 
 class TestUpdateGlobalSslSetting(object):
     def test_updates_global_hg_settings(self):
         model = VcsSettingsModel()
         with mock.patch.object(model, '_create_or_update_ui') as create_mock:
             model.update_global_ssl_setting('False')
         create_mock.assert_called_once_with(
             model.global_settings, 'web', 'push_ssl', value='False')
 
 
 class TestUpdateGlobalPathSetting(object):
     def test_updates_global_path_settings(self):
         model = VcsSettingsModel()
         with mock.patch.object(model, '_create_or_update_ui') as create_mock:
             model.update_global_path_setting('False')
         create_mock.assert_called_once_with(
             model.global_settings, 'paths', '/', value='False')
 
 
 class TestCreateOrUpdateGlobalHgSettings(object):
     FORM_DATA = {
         'extensions_largefiles': False,
         'phases_publish': False,
         'extensions_hgsubversion': False
     }
 
     def test_creates_repo_hg_settings_when_data_is_correct(self):
         model = VcsSettingsModel()
         with mock.patch.object(model, '_create_or_update_ui') as create_mock:
             model.create_or_update_global_hg_settings(self.FORM_DATA)
         expected_calls = [
             mock.call(model.global_settings, 'extensions', 'largefiles',
                       active=False, value=''),
             mock.call(model.global_settings, 'phases', 'publish',
                       value='False'),
             mock.call(model.global_settings, 'extensions', 'hgsubversion',
                       active=False)
         ]
         assert expected_calls == create_mock.call_args_list
 
     @pytest.mark.parametrize('field_to_remove', FORM_DATA.keys())
     def test_key_is_not_found(self, repo_stub, field_to_remove):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         data = self.FORM_DATA.copy()
         data.pop(field_to_remove)
         with pytest.raises(Exception) as exc_info:
             model.create_or_update_global_hg_settings(data)
         expected_message = 'The given data does not contain {} key'.format(
             field_to_remove)
         assert exc_info.value.message == expected_message
 
 
 class TestDeleteRepoSvnPattern(object):
     def test_success_when_repo_is_set(self, backend_svn):
         repo_name = backend_svn.repo_name
         model = VcsSettingsModel(repo=repo_name)
         delete_ui_patch = mock.patch.object(model.repo_settings, 'delete_ui')
         with delete_ui_patch as delete_ui_mock:
             model.delete_repo_svn_pattern(123)
         delete_ui_mock.assert_called_once_with(123)
 
     def test_raises_exception_when_repository_is_not_specified(self):
         model = VcsSettingsModel()
         with pytest.raises(Exception) as exc_info:
             model.delete_repo_svn_pattern(123)
         assert exc_info.value.message == 'Repository is not specified'
 
 
 class TestDeleteGlobalSvnPattern(object):
     def test_delete_global_svn_pattern_calls_delete_ui(self):
         model = VcsSettingsModel()
         delete_ui_patch = mock.patch.object(model.global_settings, 'delete_ui')
         with delete_ui_patch as delete_ui_mock:
             model.delete_global_svn_pattern(123)
         delete_ui_mock.assert_called_once_with(123)
 
 
 class TestFilterUiSettings(object):
     def test_settings_are_filtered(self):
         model = VcsSettingsModel()
         repo_settings = [
             UiSetting('extensions', 'largefiles', '', True),
             UiSetting('phases', 'publish', 'True', True),
             UiSetting('hooks', 'changegroup.repo_size', 'hook', True),
             UiSetting('hooks', 'changegroup.push_logger', 'hook', True),
             UiSetting('hooks', 'outgoing.pull_logger', 'hook', True),
             UiSetting(
                 'vcs_svn_branch', '84223c972204fa545ca1b22dac7bef5b68d7442d',
                 'test_branch', True),
             UiSetting(
                 'vcs_svn_tag', '84229c972204fa545ca1b22dac7bef5b68d7442d',
                 'test_tag', True),
         ]
         non_repo_settings = [
             UiSetting('test', 'outgoing.pull_logger', 'hook', True),
             UiSetting('hooks', 'test2', 'hook', True),
             UiSetting(
                 'vcs_svn_repo', '84229c972204fa545ca1b22dac7bef5b68d7442d',
                 'test_tag', True),
         ]
         settings = repo_settings + non_repo_settings
         filtered_settings = model._filter_ui_settings(settings)
         assert sorted(filtered_settings) == sorted(repo_settings)
 
 
 class TestFilterGeneralSettings(object):
     def test_settings_are_filtered(self):
         model = VcsSettingsModel()
         settings = {
             'rhodecode_abcde': 'value1',
             'rhodecode_vwxyz': 'value2',
         }
         general_settings = {
             'rhodecode_{}'.format(key): 'value'
             for key in VcsSettingsModel.GENERAL_SETTINGS
         }
         settings.update(general_settings)
 
         filtered_settings = model._filter_general_settings(general_settings)
         assert sorted(filtered_settings) == sorted(general_settings)
 
 
 class TestGetRepoUiSettings(object):
     def test_global_uis_are_returned_when_no_repo_uis_found(
             self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         result = model.get_repo_ui_settings()
         svn_sections = (
             VcsSettingsModel.SVN_TAG_SECTION,
             VcsSettingsModel.SVN_BRANCH_SECTION)
         expected_result = [
             s for s in model.global_settings.get_ui()
             if s.section not in svn_sections]
         assert sorted(result) == sorted(expected_result)
 
     def test_repo_uis_are_overriding_global_uis(
             self, repo_stub, settings_util):
         for section, key in VcsSettingsModel.HOOKS_SETTINGS:
             settings_util.create_repo_rhodecode_ui(
                 repo_stub, section, 'repo', key=key, active=False)
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         result = model.get_repo_ui_settings()
         for setting in result:
             locator = (setting.section, setting.key)
             if locator in VcsSettingsModel.HOOKS_SETTINGS:
                 assert setting.value == 'repo'
 
                 assert setting.active is False
 
     def test_global_svn_patterns_are_not_in_list(
             self, repo_stub, settings_util):
         svn_sections = (
             VcsSettingsModel.SVN_TAG_SECTION,
             VcsSettingsModel.SVN_BRANCH_SECTION)
         for section in svn_sections:
             settings_util.create_rhodecode_ui(
                 section, 'repo', key='deadbeef' + section, active=False)
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         result = model.get_repo_ui_settings()
         for setting in result:
             assert setting.section not in svn_sections
 
     def test_repo_uis_filtered_by_section_are_returned(
             self, repo_stub, settings_util):
         for section, key in VcsSettingsModel.HOOKS_SETTINGS:
             settings_util.create_repo_rhodecode_ui(
                 repo_stub, section, 'repo', key=key, active=False)
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         section, key = VcsSettingsModel.HOOKS_SETTINGS[0]
         result = model.get_repo_ui_settings(section=section)
         for setting in result:
             assert setting.section == section
 
     def test_repo_uis_filtered_by_key_are_returned(
             self, repo_stub, settings_util):
         for section, key in VcsSettingsModel.HOOKS_SETTINGS:
             settings_util.create_repo_rhodecode_ui(
                 repo_stub, section, 'repo', key=key, active=False)
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         section, key = VcsSettingsModel.HOOKS_SETTINGS[0]
         result = model.get_repo_ui_settings(key=key)
         for setting in result:
             assert setting.key == key
 
     def test_raises_exception_when_repository_is_not_specified(self):
         model = VcsSettingsModel()
         with pytest.raises(Exception) as exc_info:
             model.get_repo_ui_settings()
         assert exc_info.value.message == 'Repository is not specified'
 
 
 class TestGetRepoGeneralSettings(object):
     def test_global_settings_are_returned_when_no_repo_settings_found(
             self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         result = model.get_repo_general_settings()
         expected_result = model.global_settings.get_all_settings()
         assert sorted(result) == sorted(expected_result)
 
     def test_repo_uis_are_overriding_global_uis(
             self, repo_stub, settings_util):
         for key in VcsSettingsModel.GENERAL_SETTINGS:
             settings_util.create_repo_rhodecode_setting(
                 repo_stub, key, 'abcde', type_='unicode')
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         result = model.get_repo_ui_settings()
         for key in result:
             if key in VcsSettingsModel.GENERAL_SETTINGS:
                 assert result[key] == 'abcde'
 
     def test_raises_exception_when_repository_is_not_specified(self):
         model = VcsSettingsModel()
         with pytest.raises(Exception) as exc_info:
             model.get_repo_general_settings()
         assert exc_info.value.message == 'Repository is not specified'
 
 
 class TestGetGlobalGeneralSettings(object):
     def test_global_settings_are_returned(self, repo_stub):
         model = VcsSettingsModel()
         result = model.get_global_general_settings()
         expected_result = model.global_settings.get_all_settings()
         assert sorted(result) == sorted(expected_result)
 
     def test_repo_uis_are_not_overriding_global_uis(
             self, repo_stub, settings_util):
         for key in VcsSettingsModel.GENERAL_SETTINGS:
             settings_util.create_repo_rhodecode_setting(
                 repo_stub, key, 'abcde', type_='unicode')
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         result = model.get_global_general_settings()
         expected_result = model.global_settings.get_all_settings()
         assert sorted(result) == sorted(expected_result)
 
 
 class TestGetGlobalUiSettings(object):
     def test_global_uis_are_returned(self, repo_stub):
         model = VcsSettingsModel()
         result = model.get_global_ui_settings()
         expected_result = model.global_settings.get_ui()
         assert sorted(result) == sorted(expected_result)
 
     def test_repo_uis_are_not_overriding_global_uis(
             self, repo_stub, settings_util):
         for section, key in VcsSettingsModel.HOOKS_SETTINGS:
             settings_util.create_repo_rhodecode_ui(
                 repo_stub, section, 'repo', key=key, active=False)
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         result = model.get_global_ui_settings()
         expected_result = model.global_settings.get_ui()
         assert sorted(result) == sorted(expected_result)
 
     def test_ui_settings_filtered_by_section(
             self, repo_stub, settings_util):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         section, key = VcsSettingsModel.HOOKS_SETTINGS[0]
         result = model.get_global_ui_settings(section=section)
         expected_result = model.global_settings.get_ui(section=section)
         assert sorted(result) == sorted(expected_result)
 
     def test_ui_settings_filtered_by_key(
             self, repo_stub, settings_util):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         section, key = VcsSettingsModel.HOOKS_SETTINGS[0]
         result = model.get_global_ui_settings(key=key)
         expected_result = model.global_settings.get_ui(key=key)
         assert sorted(result) == sorted(expected_result)
 
 
 class TestGetGeneralSettings(object):
     def test_global_settings_are_returned_when_inherited_is_true(
             self, repo_stub, settings_util):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         model.inherit_global_settings = True
         for key in VcsSettingsModel.GENERAL_SETTINGS:
             settings_util.create_repo_rhodecode_setting(
                 repo_stub, key, 'abcde', type_='unicode')
         result = model.get_general_settings()
         expected_result = model.get_global_general_settings()
         assert sorted(result) == sorted(expected_result)
 
     def test_repo_settings_are_returned_when_inherited_is_false(
             self, repo_stub, settings_util):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         model.inherit_global_settings = False
         for key in VcsSettingsModel.GENERAL_SETTINGS:
             settings_util.create_repo_rhodecode_setting(
                 repo_stub, key, 'abcde', type_='unicode')
         result = model.get_general_settings()
         expected_result = model.get_repo_general_settings()
         assert sorted(result) == sorted(expected_result)
 
     def test_global_settings_are_returned_when_no_repository_specified(self):
         model = VcsSettingsModel()
         result = model.get_general_settings()
         expected_result = model.get_global_general_settings()
         assert sorted(result) == sorted(expected_result)
 
 
 class TestGetUiSettings(object):
     def test_global_settings_are_returned_when_inherited_is_true(
             self, repo_stub, settings_util):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         model.inherit_global_settings = True
         for section, key in VcsSettingsModel.HOOKS_SETTINGS:
             settings_util.create_repo_rhodecode_ui(
                 repo_stub, section, 'repo', key=key, active=True)
         result = model.get_ui_settings()
         expected_result = model.get_global_ui_settings()
         assert sorted(result) == sorted(expected_result)
 
     def test_repo_settings_are_returned_when_inherited_is_false(
             self, repo_stub, settings_util):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         model.inherit_global_settings = False
         for section, key in VcsSettingsModel.HOOKS_SETTINGS:
             settings_util.create_repo_rhodecode_ui(
                 repo_stub, section, 'repo', key=key, active=True)
         result = model.get_ui_settings()
         expected_result = model.get_repo_ui_settings()
         assert sorted(result) == sorted(expected_result)
 
     def test_repo_settings_filtered_by_section_and_key(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         model.inherit_global_settings = False
         args = ('section', 'key')
         with mock.patch.object(model, 'get_repo_ui_settings') as settings_mock:
             model.get_ui_settings(*args)
         settings_mock.assert_called_once_with(*args)
 
     def test_global_settings_filtered_by_section_and_key(self):
         model = VcsSettingsModel()
         args = ('section', 'key')
         with mock.patch.object(model, 'get_global_ui_settings') as (
                 settings_mock):
             model.get_ui_settings(*args)
         settings_mock.assert_called_once_with(*args)
 
     def test_global_settings_are_returned_when_no_repository_specified(self):
         model = VcsSettingsModel()
         result = model.get_ui_settings()
         expected_result = model.get_global_ui_settings()
         assert sorted(result) == sorted(expected_result)
 
 
 class TestGetSvnPatterns(object):
     def test_repo_settings_filtered_by_section_and_key(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub.repo_name)
         args = ('section', )
         with mock.patch.object(model, 'get_repo_ui_settings') as settings_mock:
             model.get_svn_patterns(*args)
         settings_mock.assert_called_once_with(*args)
 
     def test_global_settings_filtered_by_section_and_key(self):
         model = VcsSettingsModel()
         args = ('section', )
         with mock.patch.object(model, 'get_global_ui_settings') as (
                 settings_mock):
             model.get_svn_patterns(*args)
         settings_mock.assert_called_once_with(*args)
 
 
 class TestGetReposLocation(object):
     def test_returns_repos_location(self, repo_stub):
         model = VcsSettingsModel()
 
         result_mock = mock.Mock()
         result_mock.ui_value = '/tmp'
 
         with mock.patch.object(model, 'global_settings') as settings_mock:
             settings_mock.get_ui_by_key.return_value = result_mock
             result = model.get_repos_location()
 
         settings_mock.get_ui_by_key.assert_called_once_with('/')
         assert result == '/tmp'
 
 
 class TestCreateOrUpdateRepoSettings(object):
     FORM_DATA = {
         'inherit_global_settings': False,
         'hooks_changegroup_repo_size': False,
         'hooks_changegroup_push_logger': False,
         'hooks_outgoing_pull_logger': False,
         'extensions_largefiles': False,
         'phases_publish': 'false',
         'rhodecode_pr_merge_enabled': False,
         'rhodecode_use_outdated_comments': False,
         'new_svn_branch': '',
         'new_svn_tag': ''
     }
 
     def test_get_raises_exception_when_repository_not_specified(self):
         model = VcsSettingsModel()
         with pytest.raises(Exception) as exc_info:
             model.create_or_update_repo_settings(data=self.FORM_DATA)
         assert exc_info.value.message == 'Repository is not specified'
 
     def test_only_svn_settings_are_updated_when_type_is_svn(self, backend_svn):
         repo = backend_svn.create_repo()
         model = VcsSettingsModel(repo=repo)
         with self._patch_model(model) as mocks:
             model.create_or_update_repo_settings(
                 data=self.FORM_DATA, inherit_global_settings=False)
         mocks['create_repo_svn_settings'].assert_called_once_with(
             self.FORM_DATA)
         non_called_methods = (
             'create_or_update_repo_hook_settings',
             'create_or_update_repo_pr_settings',
             'create_or_update_repo_hg_settings')
         for method in non_called_methods:
             assert mocks[method].call_count == 0
 
     def test_non_svn_settings_are_updated_when_type_is_hg(self, backend_hg):
         repo = backend_hg.create_repo()
         model = VcsSettingsModel(repo=repo)
         with self._patch_model(model) as mocks:
             model.create_or_update_repo_settings(
                 data=self.FORM_DATA, inherit_global_settings=False)
 
         assert mocks['create_repo_svn_settings'].call_count == 0
         called_methods = (
             'create_or_update_repo_hook_settings',
             'create_or_update_repo_pr_settings',
             'create_or_update_repo_hg_settings')
         for method in called_methods:
             mocks[method].assert_called_once_with(self.FORM_DATA)
 
     def test_non_svn_and_hg_settings_are_updated_when_type_is_git(
             self, backend_git):
         repo = backend_git.create_repo()
         model = VcsSettingsModel(repo=repo)
         with self._patch_model(model) as mocks:
             model.create_or_update_repo_settings(
                 data=self.FORM_DATA, inherit_global_settings=False)
 
         assert mocks['create_repo_svn_settings'].call_count == 0
         called_methods = (
             'create_or_update_repo_hook_settings',
             'create_or_update_repo_pr_settings')
         non_called_methods = (
             'create_repo_svn_settings',
             'create_or_update_repo_hg_settings'
         )
         for method in called_methods:
             mocks[method].assert_called_once_with(self.FORM_DATA)
         for method in non_called_methods:
             assert mocks[method].call_count == 0
 
     def test_no_methods_are_called_when_settings_are_inherited(
             self, backend):
         repo = backend.create_repo()
         model = VcsSettingsModel(repo=repo)
         with self._patch_model(model) as mocks:
             model.create_or_update_repo_settings(
                 data=self.FORM_DATA, inherit_global_settings=True)
         for method_name in mocks:
             assert mocks[method_name].call_count == 0
 
     def test_cache_is_marked_for_invalidation(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub)
         invalidation_patcher = mock.patch(
             'rhodecode.controllers.admin.repos.ScmModel.mark_for_invalidation')
         with invalidation_patcher as invalidation_mock:
             model.create_or_update_repo_settings(
                 data=self.FORM_DATA, inherit_global_settings=True)
         invalidation_mock.assert_called_once_with(
             repo_stub.repo_name, delete=True)
 
     def test_inherit_flag_is_saved(self, repo_stub):
         model = VcsSettingsModel(repo=repo_stub)
         model.inherit_global_settings = True
         with self._patch_model(model):
             model.create_or_update_repo_settings(
                 data=self.FORM_DATA, inherit_global_settings=False)
         assert model.inherit_global_settings is False
 
     def _patch_model(self, model):
         return mock.patch.multiple(
             model,
             create_repo_svn_settings=mock.DEFAULT,
             create_or_update_repo_hook_settings=mock.DEFAULT,
             create_or_update_repo_pr_settings=mock.DEFAULT,
             create_or_update_repo_hg_settings=mock.DEFAULT)