repo-settings: ensure deletion on repo settings model validate

the settings id against the initialized model repo.

- prevents from malicious deletions of settings by forgin IDs

rhodecode/model/settings.py

@@ -1,814 +1,821 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import os
 import hashlib
 import logging
 from collections import namedtuple
 from functools import wraps
 
 from rhodecode.lib import caches
 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')
+    def __init__(self, setting_id):
+        msg = 'Setting `{}` is not found'.format(setting_id)
+        super(SettingNotFound, self).__init__(msg)
 
 
 class SettingsModel(BaseModel):
     BUILTIN_HOOKS = (
         RhodeCodeUi.HOOK_REPO_SIZE, RhodeCodeUi.HOOK_PUSH,
         RhodeCodeUi.HOOK_PRE_PUSH, RhodeCodeUi.HOOK_PRETX_PUSH,
         RhodeCodeUi.HOOK_PULL, RhodeCodeUi.HOOK_PRE_PULL,
         RhodeCodeUi.HOOK_PUSH_KEY,)
     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()
+            raise SettingNotFound(id_)
         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 inside the database'.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',
         'hg_use_rebase_for_merging',
         'hg_close_branch_before_merging',
         'git_use_rebase_for_merging',
         'git_close_branch_before_merging')
 
     HOOKS_SETTINGS = (
         ('hooks', 'changegroup.repo_size'),
         ('hooks', 'changegroup.push_logger'),
         ('hooks', 'outgoing.pull_logger'),)
     HG_SETTINGS = (
         ('extensions', 'largefiles'),
         ('phases', 'publish'),
         ('extensions', 'evolve'),)
     GIT_SETTINGS = (
         ('vcs_git_lfs', 'enabled'),)
     GLOBAL_HG_SETTINGS = (
         ('extensions', 'largefiles'),
         ('largefiles', 'usercache'),
         ('phases', 'publish'),
         ('extensions', 'hgsubversion'),
         ('extensions', 'evolve'),)
     GLOBAL_GIT_SETTINGS = (
         ('vcs_git_lfs', 'enabled'),
         ('vcs_git_lfs', 'store_location'))
     GLOBAL_SVN_SETTINGS = (
         ('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.GIT_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)
 
             if repo.repo_type == 'git':
                 self.create_or_update_repo_git_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)
 
     @assert_repo_settings
     def create_or_update_repo_hg_settings(self, data):
         largefiles, phases, evolve = \
             self.HG_SETTINGS
         largefiles_key, phases_key, evolve_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, *evolve, value='',
             active=data[evolve_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, largefiles_store, phases, hgsubversion, evolve \
             = self.GLOBAL_HG_SETTINGS
         largefiles_key, largefiles_store_key, phases_key, subversion_key, evolve_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, *largefiles_store,
             value=data[largefiles_store_key])
         self._create_or_update_ui(
             self.global_settings, *phases, value=safe_str(data[phases_key]))
         self._create_or_update_ui(
             self.global_settings, *hgsubversion, active=data[subversion_key])
         self._create_or_update_ui(
             self.global_settings, *evolve, value='',
             active=data[evolve_key])
 
     def create_or_update_repo_git_settings(self, data):
         # NOTE(marcink): # comma make unpack work properly
         lfs_enabled, \
             = self.GIT_SETTINGS
 
         lfs_enabled_key, \
             = self._get_settings_keys(self.GIT_SETTINGS, data)
 
         self._create_or_update_ui(
             self.repo_settings, *lfs_enabled, value=data[lfs_enabled_key],
             active=data[lfs_enabled_key])
 
     def create_or_update_global_git_settings(self, data):
         lfs_enabled, lfs_store_location \
             = self.GLOBAL_GIT_SETTINGS
         lfs_enabled_key, lfs_store_location_key \
             = self._get_settings_keys(self.GLOBAL_GIT_SETTINGS, data)
 
         self._create_or_update_ui(
             self.global_settings, *lfs_enabled, value=data[lfs_enabled_key],
             active=data[lfs_enabled_key])
         self._create_or_update_ui(
             self.global_settings, *lfs_store_location,
             value=data[lfs_store_location_key])
 
     def create_or_update_global_svn_settings(self, data):
         # 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, *http_requests_enabled,
             value=safe_str(data[http_requests_enabled_key]))
         self._create_or_update_ui(
             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_)
+        ui = self.repo_settings.UiDbModel.get(id_)
+        if ui and ui.repository.repo_name == self.repo_settings.repo:
+            # only delete if it's the same repo as initialized settings
+            self.repo_settings.delete_ui(id_)
+        else:
+            # raise error as if we wouldn't find this option
+            self.repo_settings.delete_ui(-1)
 
     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
                 elif result_key in ['vcs_git_lfs_enabled']:
                     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_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
 
     def create_largeobjects_dirs_if_needed(self, repo_store_path):
         """
         This is subscribed to the `pyramid.events.ApplicationCreated` event. It
         does a repository scan if enabled in the settings.
         """
 
         from rhodecode.lib.vcs.backends.hg import largefiles_store
         from rhodecode.lib.vcs.backends.git import lfs_store
 
         paths = [
             largefiles_store(repo_store_path),
             lfs_store(repo_store_path)]
 
         for path in paths:
             if os.path.isdir(path):
                 continue
             if os.path.isfile(path):
                 continue
             # not a file nor dir, we try to create it
             try:
                 os.makedirs(path)
             except Exception:
                 log.warning('Failed to create largefiles dir:%s', path)

rhodecode/tests/models/settings/test_settings.py

@@ -1,807 +1,807 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import mock
 import pytest
 
 from rhodecode.lib.utils2 import safe_str
 from rhodecode.model.db import (
     RhodeCodeUi, RepoRhodeCodeUi, RhodeCodeSetting, RepoRhodeCodeSetting)
 from rhodecode.model.meta import Session
 from rhodecode.model.settings import SettingsModel, SettingNotFound, UiSetting
 
 
 class TestRepoGetUiByKey(object):
     def test_ui_settings_are_returned_when_key_is_found(
             self, repo_stub, settings_util):
         section = 'test section'
         value = 'test value'
 
         settings_util.create_repo_rhodecode_ui(
             repo_stub, 'wrong section', 'wrong value')
         setting = settings_util.create_repo_rhodecode_ui(
             repo_stub, section, value)
         key = setting.ui_key
 
         model = SettingsModel(repo=repo_stub.repo_name)
         result = model.get_ui_by_key(key)
         assert result.ui_value == value
         assert result.ui_section == section
         assert result.ui_active is True
 
     def test_none_is_returned_when_key_is_not_found(
             self, repo_stub, settings_util):
         settings_util.create_repo_rhodecode_ui(
             repo_stub, 'wrong section', 'wrong value')
 
         model = SettingsModel(repo=repo_stub.repo_name)
         result = model.get_ui_by_key('abcde')
         assert result is None
 
 
 class TestGlobalGetUiByKey(object):
     def test_ui_settings_are_returned_when_key_is_found(self, settings_util):
         section = 'test section'
         value = 'test value'
 
         settings_util.create_rhodecode_ui('wrong section', 'wrong value')
         setting = settings_util.create_rhodecode_ui(section, value)
         key = setting.ui_key
 
         model = SettingsModel()
         result = model.get_ui_by_key(key)
         assert result.ui_value == value
         assert result.ui_section == section
         assert result.ui_active is True
 
     def test_none_is_returned_when_key_is_not_found(self, settings_util):
         settings_util.create_rhodecode_ui('wrong section', 'wrong value')
         model = SettingsModel()
         result = model.get_ui_by_key('abcde')
         assert result is None
 
 
 class TestRepoGetUiBySection(object):
     def test_ui_settings_are_returned_when_section_is_found(
             self, repo_stub, settings_util):
         section = 'test section'
         values = ['test value 1', 'test value 2']
 
         expected_pairs = []
         for value in values:
             setting = settings_util.create_repo_rhodecode_ui(
                 repo_stub, section, value)
             expected_pairs.append((setting.ui_key, value))
 
         model = SettingsModel(repo=repo_stub.repo_name)
         result = model.get_ui_by_section(section)
         result_pairs = [(r.ui_key, r.ui_value) for r in result]
         assert sorted(result_pairs) == sorted(expected_pairs)
 
     def test_empty_list_is_returned_when_section_is_not_found(
             self, repo_stub, settings_util):
         settings_util.create_repo_rhodecode_ui(
             repo_stub, 'wrong section', 'wrong value')
 
         model = SettingsModel(repo=repo_stub.repo_name)
         result = model.get_ui_by_section('correct section')
         assert result == []
 
 
 class TestGlobalGetUiBySection(object):
     def test_ui_settings_are_returned_when_section_is_found(
             self, settings_util):
         section = 'test section'
         values = ['test value 1', 'test value 2']
 
         expected_pairs = []
         for value in values:
             setting = settings_util.create_rhodecode_ui(section, value)
             expected_pairs.append((setting.ui_key, value))
 
         model = SettingsModel()
         result = model.get_ui_by_section(section)
         result_pairs = [(r.ui_key, r.ui_value) for r in result]
         assert sorted(result_pairs) == sorted(expected_pairs)
 
     def test_empty_list_is_returned_when_section_is_not_found(
             self, settings_util):
         settings_util.create_rhodecode_ui('wrong section', 'wrong value')
 
         model = SettingsModel()
         result = model.get_ui_by_section('correct section')
         assert result == []
 
 
 class TestRepoGetUiBySectionAndKey(object):
     def test_ui_settings_are_returned_when_section_and_key_are_found(
             self, repo_stub, settings_util):
         section = 'test section'
         value = 'test value'
         key = 'test key'
 
         settings_util.create_rhodecode_ui(
             'wrong section', 'wrong value', key='wrong key')
         setting = settings_util.create_repo_rhodecode_ui(
             repo_stub, section, value, key=key)
         key = setting.ui_key
 
         model = SettingsModel(repo=repo_stub.repo_name)
         result = model.get_ui_by_section_and_key(section, key)
         assert result.ui_value == value
         assert result.ui_section == section
         assert result.ui_active is True
 
     def test_none_is_returned_when_key_section_pair_is_not_found(
             self, repo_stub, settings_util):
         settings_util.create_repo_rhodecode_ui(
             repo_stub, 'section', 'wrong value', key='wrong key')
 
         model = SettingsModel(repo=repo_stub.repo_name)
         result = model.get_ui_by_section_and_key('section', 'test key')
         assert result is None
 
 
 class TestGlobalGetUiBySectionAndKey(object):
     def test_ui_settings_are_returned_when_section_and_key_are_found(
             self, settings_util):
         section = 'test section'
         value = 'test value'
         key = 'test key'
 
         settings_util.create_rhodecode_ui(
             'wrong section', 'wrong value', key='wrong key')
         setting = settings_util.create_rhodecode_ui(section, value, key=key)
         key = setting.ui_key
 
         model = SettingsModel()
         result = model.get_ui_by_section_and_key(section, key)
         assert result.ui_value == value
         assert result.ui_section == section
         assert result.ui_active is True
 
     def test_none_is_returned_when_key_section_pair_is_not_found(
             self, settings_util):
         settings_util.create_rhodecode_ui(
             'section', 'wrong value', key='wrong key')
         model = SettingsModel()
         result = model.get_ui_by_section_and_key('section', 'test key')
         assert result is None
 
 
 class TestRepoGetUi(object):
     def test_non_empty_list_is_returned_when_ui_settings_found(
             self, repo_stub, settings_util, fake_ui_values):
         for ui in fake_ui_values:
             settings_util.create_repo_rhodecode_ui(
                 repo_stub, ui.section, ui.value, key=ui.key)
             # Create few global settings to check that only repo ones are
             # displayed
             settings_util.create_rhodecode_ui(ui.section, ui.value, key=ui.key)
 
         model = SettingsModel(repo=repo_stub.repo_name)
         result = model.get_ui()
         assert sorted(result) == sorted(fake_ui_values)
 
     def test_settings_filtered_by_section(
             self, repo_stub, settings_util, fake_ui_values):
         for ui in fake_ui_values:
             settings_util.create_repo_rhodecode_ui(
                 repo_stub, ui.section, ui.value, key=ui.key)
 
         model = SettingsModel(repo=repo_stub.repo_name)
         result = model.get_ui(section=fake_ui_values[0].section)
         expected_result = [
             s for s in fake_ui_values
             if s.section == fake_ui_values[0].section]
         assert sorted(result) == sorted(expected_result)
 
     def test_settings_filtered_by_key(
             self, repo_stub, settings_util, fake_ui_values):
         for ui in fake_ui_values:
             settings_util.create_repo_rhodecode_ui(
                 repo_stub, ui.section, ui.value, key=ui.key)
 
         model = SettingsModel(repo=repo_stub.repo_name)
         result = model.get_ui(key=fake_ui_values[0].key)
         expected_result = [
             s for s in fake_ui_values if s.key == fake_ui_values[0].key]
         assert sorted(result) == sorted(expected_result)
 
     def test_empty_list_is_returned_when_ui_settings_are_not_found(
             self, repo_stub, settings_util):
         for i in range(10):
             settings_util.create_rhodecode_ui(
                 'section{}'.format(i), 'value{}'.format(i),
                 key='key{}'.format(i), active=True)
 
         model = SettingsModel(repo=repo_stub.repo_name)
         result = model.get_ui()
         assert result == []
 
 
 class TestGlobalGetUi(object):
     def test_non_empty_list_is_returned_when_ui_settings_found(
             self, backend_stub, settings_util, fake_ui_values):
         repo = backend_stub.create_repo()
         for ui in fake_ui_values:
             settings_util.create_rhodecode_ui(ui.section, ui.value, key=ui.key)
             # Create few repo settings to check that only global ones are
             # displayed
             settings_util.create_repo_rhodecode_ui(
                 repo, ui.section, ui.value, key=ui.key)
 
         model = SettingsModel()
         result = model.get_ui()
         for ui in fake_ui_values:
             assert ui in result
 
     def test_settings_filtered_by_key(self, settings_util, fake_ui_values):
         for ui in fake_ui_values:
             settings_util.create_rhodecode_ui(ui.section, ui.value, key=ui.key)
         expected_result = [
             s for s in fake_ui_values if s.key == fake_ui_values[0].key]
 
         model = SettingsModel()
         result = model.get_ui(key=fake_ui_values[0].key)
         assert sorted(result) == sorted(expected_result)
 
     def test_settings_filtered_by_section(self, settings_util, fake_ui_values):
         for ui in fake_ui_values:
             settings_util.create_rhodecode_ui(ui.section, ui.value, key=ui.key)
         expected_result = [
             s for s in fake_ui_values
             if s.section == fake_ui_values[0].section]
 
         model = SettingsModel()
         result = model.get_ui(section=fake_ui_values[0].section)
         assert sorted(result) == sorted(expected_result)
 
     def test_repo_settings_are_not_displayed(
             self, backend_stub, settings_util, fake_ui_values):
         repo = backend_stub.create_repo()
         for ui in fake_ui_values:
             settings_util.create_repo_rhodecode_ui(
                 repo, ui.section, ui.value, key=ui.key, active=ui.active)
 
         model = SettingsModel()
         result = model.get_ui()
         for ui in fake_ui_values:
             assert ui not in result
 
 
 class TestRepoGetBuiltInHooks(object):
     def test_only_builtin_hooks_are_returned(self, repo_stub, settings_util):
         section = 'hooks'
         valid_keys = SettingsModel.BUILTIN_HOOKS
         invalid_keys = ('fake_hook', )
         keys = valid_keys + invalid_keys
 
         for key in keys:
             settings_util.create_repo_rhodecode_ui(
                 repo_stub, section, 'test value', key=key)
 
         model = SettingsModel(repo=repo_stub.repo_name)
         result = model.get_builtin_hooks()
 
         assert len(result) == len(valid_keys)
         for entry in result:
             assert entry.ui_key in valid_keys
 
 
 class TestGlobalGetBuiltInHooks(object):
     def test_only_builtin_hooks_are_returned(self, settings_util):
         section = 'hooks'
         valid_keys = ('valid_key1', 'valid_key2')
         invalid_keys = ('fake_hook', )
         keys = valid_keys + invalid_keys
 
         for key in keys:
             settings_util.create_rhodecode_ui(section, 'test value', key=key)
 
         model = SettingsModel()
         with mock.patch.object(model, 'BUILTIN_HOOKS', valid_keys):
             result = model.get_builtin_hooks()
 
         assert len(result) == len(valid_keys)
         for entry in result:
             assert entry.ui_key in valid_keys
 
 
 class TestRepoGetCustomHooks(object):
     def test_only_custom_hooks_are_returned(self, repo_stub, settings_util):
         section = 'hooks'
         valid_keys = ('custom', )
         invalid_keys = SettingsModel.BUILTIN_HOOKS
         keys = valid_keys + invalid_keys
 
         for key in keys:
             settings_util.create_repo_rhodecode_ui(
                 repo_stub, section, 'test value', key=key)
 
         model = SettingsModel(repo=repo_stub.repo_name)
         result = model.get_custom_hooks()
 
         assert len(result) == len(valid_keys)
         for entry in result:
             assert entry.ui_key in valid_keys
 
 
 class TestGlobalGetCustomHooks(object):
     def test_only_custom_hooks_are_returned(self, settings_util):
         section = 'hooks'
         valid_keys = ('valid_key1', 'valid_key2')
         invalid_keys = ('fake_hook', )
         keys = valid_keys + invalid_keys
 
         for key in keys:
             settings_util.create_rhodecode_ui(section, 'test value', key=key)
 
         model = SettingsModel()
         with mock.patch.object(model, 'BUILTIN_HOOKS', invalid_keys):
             result = model.get_custom_hooks()
         for entry in result:
             assert entry.ui_key not in invalid_keys
 
 
 class TestRepoCreateUiSectionValue(object):
     @pytest.mark.parametrize("additional_kwargs", [
         {'key': 'abcde'},
         {'active': False},
         {}
     ])
     def test_ui_section_value_is_created(
             self, repo_stub, additional_kwargs):
         model = SettingsModel(repo=repo_stub.repo_name)
         section = 'test section'
         value = 'test value'
         result = model.create_ui_section_value(section, value)
         key = result.ui_key
         Session().commit()
 
         setting = model.get_ui_by_key(key)
         try:
             assert setting == result
             assert isinstance(setting, RepoRhodeCodeUi)
         finally:
             Session().delete(result)
             Session().commit()
 
 
 class TestGlobalCreateUiSectionValue(object):
     @pytest.mark.parametrize("additional_kwargs", [
         {'key': 'abcde'},
         {'active': False},
         {}
     ])
     def test_ui_section_value_is_created_with_autogenerated_key(
             self, backend_stub, additional_kwargs):
         model = SettingsModel()
         section = 'test section'
         value = 'test value'
         result = model.create_ui_section_value(
             section, value, **additional_kwargs)
         key = result.ui_key
         Session().commit()
 
         setting = model.get_ui_by_key(key)
         try:
             assert setting == result
             assert isinstance(setting, RhodeCodeUi)
         finally:
             Session().delete(result)
             Session().commit()
 
 
 class TestRepoCreateOrUpdateHook(object):
     def test_hook_created(self, repo_stub):
         model = SettingsModel(repo=repo_stub.repo_name)
         key = 'test_key'
         value = 'test value'
         result = model.create_or_update_hook(key, value)
         Session().commit()
 
         setting = model.get_ui_by_section_and_key('hooks', key)
         try:
             assert setting == result
             assert isinstance(setting, RepoRhodeCodeUi)
         finally:
             Session().delete(result)
             Session().commit()
 
     def test_hook_updated(self, repo_stub, settings_util):
         section = 'hooks'
         key = 'test_key'
 
         settings_util.create_repo_rhodecode_ui(
             repo_stub, section, 'old value', key=key)
 
         model = SettingsModel(repo=repo_stub.repo_name)
         value = 'test value'
         model.create_or_update_hook(key, value)
         Session().commit()
 
         setting = model.get_ui_by_section_and_key('hooks', key)
         assert setting.ui_value == value
 
 
 class TestGlobalCreateOrUpdateHook(object):
     def test_hook_created(self):
         model = SettingsModel()
         key = 'test_key'
         value = 'test value'
         result = model.create_or_update_hook(key, value)
         Session().commit()
 
         setting = model.get_ui_by_section_and_key('hooks', key)
         try:
             assert setting == result
             assert isinstance(setting, RhodeCodeUi)
         finally:
             Session().delete(result)
             Session().commit()
 
     def test_hook_updated(self, settings_util):
         section = 'hooks'
         key = 'test_key'
 
         settings_util.create_rhodecode_ui(section, 'old value', key=key)
 
         model = SettingsModel()
         value = 'test value'
         model.create_or_update_hook(key, value)
         Session().commit()
 
         setting = model.get_ui_by_section_and_key('hooks', key)
         assert setting.ui_value == value
 
 
 class TestDeleteUiValue(object):
     def test_delete_ui_when_repo_is_set(self, repo_stub, settings_util):
         model = SettingsModel(repo=repo_stub.repo_name)
         result = settings_util.create_repo_rhodecode_ui(
             repo_stub, 'section', None, cleanup=False)
 
         key = result.ui_key
         model.delete_ui(result.ui_id)
         Session().commit()
 
         setting = model.get_ui_by_key(key)
         assert setting is None
 
     @pytest.mark.parametrize('id_', (None, 123))
     def test_raises_exception_when_id_is_not_specified(self, id_):
         model = SettingsModel()
         with pytest.raises(SettingNotFound) as exc_info:
             model.delete_ui(id_)
-        assert exc_info.value.message == 'Setting is not found'
+        assert exc_info.value.message == 'Setting `{}` is not found'.format(id_)
 
     def test_delete_ui_when_repo_is_not_set(self, settings_util):
         model = SettingsModel()
         result = settings_util.create_rhodecode_ui(
             'section', None, cleanup=False)
 
         key = result.ui_key
         model.delete_ui(result.ui_id)
         Session().commit()
 
         setting = model.get_ui_by_key(key)
         assert setting is None
 
 
 class TestRepoGetSettingByName(object):
     @pytest.mark.parametrize("name, value, type_, expected_value", [
         ('test_unicode', 'Straße', 'unicode', 'Straße'),
         ('test_int', '1234', 'int', 1234),
         ('test_bool', 'True', 'bool', True),
         ('test_list', 'a,b,c', 'list', ['a', 'b', 'c'])
     ])
     def test_setting_is_returned_when_name_is_found(
             self, repo_stub, settings_util, name, value, type_,
             expected_value):
         settings_util.create_repo_rhodecode_setting(
             repo_stub, name, value, type_)
 
         model = SettingsModel(repo=repo_stub.repo_name)
         setting = model.get_setting_by_name(name)
         assert setting.app_settings_type == type_
         actual_value = setting.app_settings_value
         if type_ == 'unicode':
             actual_value = safe_str(actual_value)
         assert actual_value == expected_value
 
     def test_returns_none_if_the_setting_does_not_exist(self, repo_stub):
         model = SettingsModel(repo=repo_stub.repo_name)
         setting = model.get_setting_by_name('abcde')
         assert setting is None
 
 
 class TestGlobalGetSettingByName(object):
     @pytest.mark.parametrize("name, value, type_, expected_value", [
         ('test_unicode', 'Straße', 'unicode', 'Straße'),
         ('test_int', '1234', 'int', 1234),
         ('test_bool', 'True', 'bool', True),
         ('test_list', 'a,b,c', 'list', ['a', 'b', 'c'])
     ])
     def test_setting_is_returned_when_name_is_found(
             self, settings_util, name, value, type_, expected_value):
         settings_util.create_rhodecode_setting(name, value, type_)
 
         model = SettingsModel()
         setting = model.get_setting_by_name(name)
         assert setting.app_settings_type == type_
         actual_value = setting.app_settings_value
         if type_ == 'unicode':
             actual_value = safe_str(actual_value)
         assert actual_value == expected_value
 
     def test_returns_none_if_the_setting_does_not_exist(self):
         model = SettingsModel()
         setting = model.get_setting_by_name('abcde')
         assert setting is None
 
 
 class TestRepoGetAllSettings(object):
     def test_settings_are_found(self, repo_stub, settings_util):
         initial_settings = {
             'test_setting_{}'.format(i): 'value' for i in range(10)}
         settings = [
             settings_util.create_repo_rhodecode_setting(
                 repo_stub, name, initial_settings[name], 'unicode')
             for name in initial_settings
         ]
         model = SettingsModel(repo=repo_stub.repo_name)
 
         settings = model.get_all_settings()
         expected_settings = {
             'rhodecode_' + name: initial_settings[name]
             for name in initial_settings
         }
 
         assert len(settings) == 10
         assert expected_settings == settings
 
     def test_settings_are_not_found(self, repo_stub):
         model = SettingsModel(repo=repo_stub.repo_name)
         setting = model.get_all_settings()
         assert setting == {}
 
 
 class TestGlobalGetAllSettings(object):
     def test_settings_are_found(self, settings_util):
         initial_settings = {
             'test_setting_{}'.format(i): 'value' for i in range(10)}
         settings = [
             settings_util.create_rhodecode_setting(
                 name, initial_settings[name], 'unicode')
             for name in initial_settings
         ]
         model = SettingsModel()
 
         settings = model.get_all_settings()
         expected_settings = {
             'rhodecode_' + name: initial_settings[name]
             for name in initial_settings
         }
 
         filtered_settings = {
             name: settings[name]
             for name in settings if name.startswith('rhodecode_test_setting')
         }
         assert len(filtered_settings) == 10
         assert expected_settings == filtered_settings
 
     def test_settings_are_not_found(self, repo_stub):
         model = SettingsModel(repo=repo_stub.repo_name)
         setting = model.get_all_settings()
         assert setting == {}
 
 
 class TestRepoCreateOrUpdateSetting(object):
     def test_setting_is_created(self, repo_stub):
         model = SettingsModel(repo=repo_stub.repo_name)
         name = 'test_setting'
         value = 'test_value'
         model.create_or_update_setting(name, val=value)
 
         setting = model.get_setting_by_name(name)
         try:
             assert setting.app_settings_name == name
             assert setting.app_settings_value == value
             assert setting.app_settings_type == 'unicode'
             assert isinstance(setting, RepoRhodeCodeSetting)
         finally:
             Session().delete(setting)
             Session().commit()
 
     def test_setting_is_updated(self, repo_stub, settings_util):
         model = SettingsModel(repo=repo_stub.repo_name)
         name = 'test_setting'
         value = 'test_value'
         settings_util.create_repo_rhodecode_setting(
             repo_stub, name, value, 'unicode', cleanup=False)
 
         updated_value = 'test_value_2'
         model.create_or_update_setting(name, val=updated_value)
 
         setting = model.get_setting_by_name(name)
         try:
             assert setting.app_settings_name == name
             assert setting.app_settings_value == updated_value
             assert setting.app_settings_type == 'unicode'
             assert isinstance(setting, RepoRhodeCodeSetting)
         finally:
             Session().delete(setting)
             Session().commit()
 
 
 class TestGlobalCreateOrUpdateSetting(object):
     def test_setting_is_created(self):
         model = SettingsModel()
         name = 'test_setting'
         value = 'test_value'
         model.create_or_update_setting(name, val=value)
 
         setting = model.get_setting_by_name(name)
         try:
             assert setting.app_settings_name == name
             assert setting.app_settings_value == value
             assert setting.app_settings_type == 'unicode'
             assert isinstance(setting, RhodeCodeSetting)
         finally:
             Session().delete(setting)
             Session().commit()
 
     def test_setting_is_updated(self, settings_util):
         model = SettingsModel()
         name = 'test_setting'
         value = 'test_value'
         settings_util.create_rhodecode_setting(
             name, value, 'unicode', cleanup=False)
 
         updated_value = 'test_value_2'
         model.create_or_update_setting(name, val=updated_value)
 
         setting = model.get_setting_by_name(name)
         try:
             assert setting.app_settings_name == name
             assert setting.app_settings_value == updated_value
             assert setting.app_settings_type == 'unicode'
             assert isinstance(setting, RhodeCodeSetting)
         finally:
             Session().delete(setting)
             Session().commit()
 
 
 class TestRepoGetAuthSettings(object):
     def test_settings_prefixed_with_auth_are_retured(
             self, repo_stub, settings_util):
         model = SettingsModel(repo=repo_stub.repo_name)
         valid_settings = ('auth_test1', 'auth_test2')
         invalid_settings = ('test1', 'test2')
         fake_value = 'test_value'
 
         for name in valid_settings + invalid_settings:
             settings_util.create_repo_rhodecode_setting(
                 repo_stub, name, fake_value, 'unicode')
 
         auth_settings = model.get_auth_settings()
         assert auth_settings == {name: fake_value for name in valid_settings}
 
 
 class TestGlobalGetAuthSettings(object):
     def test_settings_prefixed_with_auth_are_retured(self, settings_util):
         model = SettingsModel()
         valid_settings = ('auth_test1', 'auth_test2')
         invalid_settings = ('test1', 'test2')
         fake_value = 'test_value'
 
         for name in valid_settings + invalid_settings:
             settings_util.create_rhodecode_setting(name, fake_value, 'unicode')
 
         auth_settings = model.get_auth_settings()
         for name in auth_settings:
             assert name not in invalid_settings
             if name in valid_settings:
                 assert auth_settings[name] == fake_value
 
 
 class TestGetAuthPlugins(object):
     def test_get_setting_by_name_is_called(self):
         model = SettingsModel()
 
         fake_value = 'some value'
         result_mock = mock.Mock()
         result_mock.app_settings_value = fake_value
 
         get_setting_patch = mock.patch.object(
             model, 'get_setting_by_name', return_value=result_mock)
 
         with get_setting_patch as get_setting_mock:
             result = model.get_auth_plugins()
 
         get_setting_mock.assert_called_once_with('auth_plugins')
         assert result == fake_value
 
 
 class TestDefaultRepoSettings(object):
     DEFAULT_SETTINGS_NAMES = ['default_a{}'.format(i) for i in range(10)]
     CUSTOM_SETTINGS_NAMES = ['setting_b_{}'.format(i) for i in range(10)]
 
     def test_returns_global_settings_prefixed_with_default(
             self, settings_util):
         self._create_values(settings_util)
         model = SettingsModel()
         result = model.get_default_repo_settings()
         self._assert_prefixed_settings(result)
 
     def test_returns_global_settings_without_default_prefix(
             self, settings_util):
         self._create_values(settings_util)
         model = SettingsModel()
         result = model.get_default_repo_settings(strip_prefix=True)
         self._assert_non_prefixed_settings(result)
 
     def test_returns_per_repo_settings_prefixed_with_default(
             self, repo_stub, settings_util):
         model = SettingsModel(repo=repo_stub)
         self._create_values(settings_util, repo=repo_stub)
         result = model.get_default_repo_settings()
         self._assert_prefixed_settings(result)
 
     def test_returns_per_repo_settings_without_default_prefix(
             self, repo_stub, settings_util):
         model = SettingsModel(repo=repo_stub)
         self._create_values(settings_util, repo=repo_stub)
         result = model.get_default_repo_settings(strip_prefix=True)
         self._assert_non_prefixed_settings(result)
 
     def _create_values(self, settings_util, repo=None):
         for name in self.DEFAULT_SETTINGS_NAMES + self.CUSTOM_SETTINGS_NAMES:
             if not repo:
                 settings_util.create_rhodecode_setting(
                     name, 'value', 'unicode')
             else:
                 settings_util.create_repo_rhodecode_setting(
                     repo, name, 'value', 'unicode')
 
     def _assert_prefixed_settings(self, result):
         for setting in self.DEFAULT_SETTINGS_NAMES:
             assert setting in result
             assert result[setting] == 'value'
 
         for setting in self.CUSTOM_SETTINGS_NAMES:
             assert setting not in result
 
     def _assert_non_prefixed_settings(self, result):
         for setting in self.DEFAULT_SETTINGS_NAMES:
             setting = setting.replace('default_', '')
             assert setting in result
             assert result[setting] == 'value'
 
 
 @pytest.fixture
 def fake_ui_values():
     return [
         UiSetting(
             'section{}'.format(i % 2), 'key{}'.format(i),
             'value{}'.format(i), True)
         for i in range(10)
     ]

rhodecode/tests/models/settings/test_vcs_settings.py

@@ -1,1072 +1,1083 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import 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,
     'rhodecode_hg_close_branch_before_merging': True,
     'rhodecode_git_use_rebase_for_merging': True,
     'rhodecode_git_close_branch_before_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.GIT_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('.', '_'))
 
             if section in ('extensions', 'hooks'):
                 value = True
             elif key in ['vcs_git_lfs_enabled']:
                 value = True
             else:
                 value = '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 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, evolve = 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,
         'extensions_evolve': 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, 'extensions', 'evolve',
                       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,
         'largefiles_usercache': '/example/largefiles-store',
         'phases_publish': False,
         'extensions_hgsubversion': False,
         'extensions_evolve': 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, 'largefiles', 'usercache',
                       value='/example/largefiles-store'),
             mock.call(model.global_settings, 'phases', 'publish',
                       value='False'),
             mock.call(model.global_settings, 'extensions', 'hgsubversion',
                       active=False),
             mock.call(model.global_settings, 'extensions', 'evolve',
                       active=False, value='')
         ]
         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 TestCreateOrUpdateGlobalGitSettings(object):
     FORM_DATA = {
         'vcs_git_lfs_enabled': False,
         'vcs_git_lfs_store_location': '/example/lfs-store',
     }
 
     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_git_settings(self.FORM_DATA)
         expected_calls = [
             mock.call(model.global_settings, 'vcs_git_lfs', 'enabled',
                       active=False, value=False),
             mock.call(model.global_settings, 'vcs_git_lfs', 'store_location',
                       value='/example/lfs-store'),
         ]
         assert expected_calls == create_mock.call_args_list
 
 
 class TestDeleteRepoSvnPattern(object):
-    def test_success_when_repo_is_set(self, backend_svn):
+    def test_success_when_repo_is_set(self, backend_svn, settings_util):
+        repo = backend_svn.create_repo()
+        repo_name = repo.repo_name
+
+        model = VcsSettingsModel(repo=repo_name)
+        entry = settings_util.create_repo_rhodecode_ui(
+            repo, VcsSettingsModel.SVN_BRANCH_SECTION, 'svn-branch')
+        Session().commit()
+
+        model.delete_repo_svn_pattern(entry.ui_id)
+
+    def test_fail_when_delete_id_from_other_repo(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)
+        delete_ui_mock.assert_called_once_with(-1)
 
     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('largefiles', 'usercache', '/example/largefiles-store', True),
             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,
         'extensions_evolve': False,
         'largefiles_usercache': '/example/largefiles-store',
         'vcs_git_lfs_enabled': False,
         'vcs_git_lfs_store_location': '/',
         '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.model.scm.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)