tests: fixed tests for new repo settings

- added settings update missing tests
- fixed id generation to some tests for better display

rhodecode/apps/repository/tests/test_repo_settings.py

@@ -0,0 +1,230 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.lib.vcs.exceptions import RepositoryRequirementError
+from rhodecode.model.db import Repository, UserRepoToPerm, Permission, User
+from rhodecode.model.meta import Session
+from rhodecode.tests import (
+    url, HG_REPO, TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN,
+    assert_session_flash)
+from rhodecode.tests.fixture import Fixture
+
+fixture = Fixture()
+
+
+def route_path(name, params=None, **kwargs):
+    import urllib
+
+    base_url = {
+        'edit_repo': '/{repo_name}/settings',
+    }[name].format(**kwargs)
+
+    if params:
+        base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
+    return base_url
+
+
+def _get_permission_for_user(user, repo):
+    perm = UserRepoToPerm.query()\
+        .filter(UserRepoToPerm.repository ==
+                Repository.get_by_repo_name(repo))\
+        .filter(UserRepoToPerm.user == User.get_by_username(user))\
+        .all()
+    return perm
+
+
[e-mail]('autologin_user', 'app')
+class TestAdminRepoSettings(object):
+    @pytest.mark.parametrize('urlname', [
+        'edit_repo',
+    ])
+    def test_show_page(self, urlname, app, backend):
+        app.get(route_path(urlname, repo_name=backend.repo_name), status=200)
+
+    def test_edit_accessible_when_missing_requirements(
+            self, backend_hg, autologin_user):
+        scm_patcher = mock.patch.object(
+            Repository, 'scm_instance', side_effect=RepositoryRequirementError)
+        with scm_patcher:
+            self.app.get(route_path('edit_repo', repo_name=backend_hg.repo_name))
+
+    @pytest.mark.parametrize('urlname', [
+        'edit_repo_perms',
+        'edit_repo_advanced',
+        'repo_vcs_settings',
+        'edit_repo_fields',
+        'repo_settings_issuetracker',
+        'edit_repo_caches',
+        'edit_repo_remote',
+        'edit_repo_statistics',
+    ])
+    def test_show_page_pylons(self, urlname, app):
+        app.get(url(urlname, repo_name=HG_REPO))
+
+    @pytest.mark.parametrize('update_settings', [
+        {'repo_description': 'alter-desc'},
+        {'repo_owner': TEST_USER_REGULAR_LOGIN},
+        {'repo_private': 'true'},
+        {'repo_enable_locking': 'true'},
+        {'repo_enable_downloads': 'true'},
+    ])
+    def test_update_repo_settings(self, update_settings, csrf_token, backend, user_util):
+        repo = user_util.create_repo(repo_type=backend.alias)
+        repo_name = repo.repo_name
+
+        params = fixture._get_repo_create_params(
+                csrf_token=csrf_token,
+                repo_name=repo_name,
+                repo_type=backend.alias,
+                repo_owner=TEST_USER_ADMIN_LOGIN,
+                repo_description='DESC',
+
+                repo_private='false',
+                repo_enable_locking='false',
+                repo_enable_downloads='false')
+        params.update(update_settings)
+        self.app.post(
+            route_path('edit_repo', repo_name=repo_name),
+            params=params, status=302)
+
+        repo = Repository.get_by_repo_name(repo_name)
+        assert repo.user.username == \
+               update_settings.get('repo_owner', repo.user.username)
+
+        assert repo.description == \
+               update_settings.get('repo_description', repo.description)
+
+        assert repo.private == \
+               str2bool(update_settings.get(
+                   'repo_private', repo.private))
+
+        assert repo.enable_locking == \
+               str2bool(update_settings.get(
+                   'repo_enable_locking', repo.enable_locking))
+
+        assert repo.enable_downloads == \
+               str2bool(update_settings.get(
+                   'repo_enable_downloads', repo.enable_downloads))
+
+    def test_update_repo_name_via_settings(self, csrf_token, user_util, backend):
+        repo = user_util.create_repo(repo_type=backend.alias)
+        repo_name = repo.repo_name
+
+        repo_group = user_util.create_repo_group()
+        repo_group_name = repo_group.group_name
+        new_name = repo_group_name + '_' + repo_name
+
+        params = fixture._get_repo_create_params(
+                csrf_token=csrf_token,
+                repo_name=new_name,
+                repo_type=backend.alias,
+                repo_owner=TEST_USER_ADMIN_LOGIN,
+                repo_description='DESC',
+                repo_private='false',
+                repo_enable_locking='false',
+                repo_enable_downloads='false')
+        self.app.post(
+            route_path('edit_repo', repo_name=repo_name),
+            params=params, status=302)
+        repo = Repository.get_by_repo_name(new_name)
+        assert repo.repo_name == new_name
+
+    def test_update_repo_group_via_settings(self, csrf_token, user_util, backend):
+        repo = user_util.create_repo(repo_type=backend.alias)
+        repo_name = repo.repo_name
+
+        repo_group = user_util.create_repo_group()
+        repo_group_name = repo_group.group_name
+        repo_group_id = repo_group.group_id
+
+        new_name = repo_group_name + '/' + repo_name
+        params = fixture._get_repo_create_params(
+                csrf_token=csrf_token,
+                repo_name=repo_name,
+                repo_type=backend.alias,
+                repo_owner=TEST_USER_ADMIN_LOGIN,
+                repo_description='DESC',
+                repo_group=repo_group_id,
+                repo_private='false',
+                repo_enable_locking='false',
+                repo_enable_downloads='false')
+        self.app.post(
+            route_path('edit_repo', repo_name=repo_name),
+            params=params, status=302)
+        repo = Repository.get_by_repo_name(new_name)
+        assert repo.repo_name == new_name
+
+    def test_set_private_flag_sets_default_user_permissions_to_none(
+            self, autologin_user, backend, csrf_token):
+
+        # initially repository perm should be read
+        perm = _get_permission_for_user(user='default', repo=backend.repo_name)
+        assert len(perm) == 1
+        assert perm[0].permission.permission_name == 'repository.read'
+        assert not backend.repo.private
+
+        response = self.app.post(
+            route_path('edit_repo', repo_name=backend.repo_name),
+            params=fixture._get_repo_create_params(
+                repo_private='true',
+                repo_name=backend.repo_name,
+                repo_type=backend.alias,
+                repo_owner=TEST_USER_ADMIN_LOGIN,
+                csrf_token=csrf_token), status=302)
+
+        assert_session_flash(
+            response,
+            msg='Repository %s updated successfully' % (backend.repo_name))
+
+        repo = Repository.get_by_repo_name(backend.repo_name)
+        assert repo.private is True
+
+        # now the repo default permission should be None
+        perm = _get_permission_for_user(user='default', repo=backend.repo_name)
+        assert len(perm) == 1
+        assert perm[0].permission.permission_name == 'repository.none'
+
+        response = self.app.post(
+            route_path('edit_repo', repo_name=backend.repo_name),
+            params=fixture._get_repo_create_params(
+                repo_private='false',
+                repo_name=backend.repo_name,
+                repo_type=backend.alias,
+                repo_owner=TEST_USER_ADMIN_LOGIN,
+                csrf_token=csrf_token), status=302)
+
+        assert_session_flash(
+            response,
+            msg='Repository %s updated successfully' % (backend.repo_name))
+        assert backend.repo.private is False
+
+        # we turn off private now the repo default permission should stay None
+        perm = _get_permission_for_user(user='default', repo=backend.repo_name)
+        assert len(perm) == 1
+        assert perm[0].permission.permission_name == 'repository.none'
+
+        # update this permission back
+        perm[0].permission = Permission.get_by_key('repository.read')
+        Session().add(perm[0])
+        Session().commit()

rhodecode/apps/repository/tests/test_vcs_settings.py

@@ -0,0 +1,117 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
+
+import rhodecode
+from rhodecode.model.settings import SettingsModel
+from rhodecode.tests import url
+from rhodecode.tests.utils import AssertResponse
+
+
+def route_path(name, params=None, **kwargs):
+    import urllib
+
+    base_url = {
+        'edit_repo': '/{repo_name}/settings',
+    }[name].format(**kwargs)
+
+    if params:
+        base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
+    return base_url
+
+
[e-mail]('autologin_user', 'app')
+class TestAdminRepoVcsSettings(object):
+
+    @pytest.mark.parametrize('setting_name, setting_backends', [
+        ('hg_use_rebase_for_merging', ['hg']),
+    ])
+    def test_labs_settings_visible_if_enabled(
+            self, setting_name, setting_backends, backend):
+        if backend.alias not in setting_backends:
+            pytest.skip('Setting not available for backend {}'.format(backend))
+
+        vcs_settings_url = url(
+            'repo_vcs_settings', repo_name=backend.repo.repo_name)
+
+        with mock.patch.dict(
+                rhodecode.CONFIG, {'labs_settings_active': 'true'}):
+            response = self.app.get(vcs_settings_url)
+
+        assertr = AssertResponse(response)
+        assertr.one_element_exists('#rhodecode_{}'.format(setting_name))
+
+    @pytest.mark.parametrize('setting_name, setting_backends', [
+        ('hg_use_rebase_for_merging', ['hg']),
+    ])
+    def test_labs_settings_not_visible_if_disabled(
+            self, setting_name, setting_backends, backend):
+        if backend.alias not in setting_backends:
+            pytest.skip('Setting not available for backend {}'.format(backend))
+
+        vcs_settings_url = url(
+            'repo_vcs_settings', repo_name=backend.repo.repo_name)
+
+        with mock.patch.dict(
+                rhodecode.CONFIG, {'labs_settings_active': 'false'}):
+            response = self.app.get(vcs_settings_url)
+
+        assertr = AssertResponse(response)
+        assertr.no_element_exists('#rhodecode_{}'.format(setting_name))
+
+    @pytest.mark.parametrize('setting_name, setting_backends', [
+        ('hg_use_rebase_for_merging', ['hg']),
+    ])
+    def test_update_boolean_settings(
+            self, csrf_token, setting_name, setting_backends, backend):
+        if backend.alias not in setting_backends:
+            pytest.skip('Setting not available for backend {}'.format(backend))
+
+        repo = backend.create_repo()
+
+        settings_model = SettingsModel(repo=repo)
+        vcs_settings_url = url(
+            'repo_vcs_settings', repo_name=repo.repo_name)
+
+        self.app.post(
+            vcs_settings_url,
+            params={
+                'inherit_global_settings': False,
+                'new_svn_branch': 'dummy-value-for-testing',
+                'new_svn_tag': 'dummy-value-for-testing',
+                'rhodecode_{}'.format(setting_name): 'true',
+                'csrf_token': csrf_token,
+            })
+        setting = settings_model.get_setting_by_name(setting_name)
+        assert setting.app_settings_value
+
+        self.app.post(
+            vcs_settings_url,
+            params={
+                'inherit_global_settings': False,
+                'new_svn_branch': 'dummy-value-for-testing',
+                'new_svn_tag': 'dummy-value-for-testing',
+                'rhodecode_{}'.format(setting_name): 'false',
+                'csrf_token': csrf_token,
+            })
+        setting = settings_model.get_setting_by_name(setting_name)
+        assert not setting.app_settings_value

rhodecode/tests/config/test_sanitize_settings.py

@@ -1,203 +1,204 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
 import pytest
 
+from rhodecode.tests import no_newline_id_generator
 from rhodecode.config.middleware import (
     _sanitize_vcs_settings, _bool_setting, _string_setting, _list_setting,
     _int_setting)
 
 
 class TestHelperFunctions(object):
     @pytest.mark.parametrize('raw, expected', [
         ('true', True), (u'true', True),
         ('yes', True), (u'yes', True),
         ('on', True), (u'on', True),
         ('false', False), (u'false', False),
         ('no', False), (u'no', False),
         ('off', False), (u'off', False),
         ('invalid-bool-value', False),
         ('invalid-∫øø@-√å@¨€', False),
         (u'invalid-∫øø@-√å@¨€', False),
     ])
     def test_bool_setting_helper(self, raw, expected):
         key = 'dummy-key'
         settings = {key: raw}
         _bool_setting(settings, key, None)
         assert settings[key] is expected
 
     @pytest.mark.parametrize('raw, expected', [
         ('', ''),
         ('test-string', 'test-string'),
         ('CaSe-TeSt', 'case-test'),
         ('test-string-烩€', 'test-string-烩€'),
         (u'test-string-烩€', u'test-string-烩€'),
     ])
     def test_string_setting_helper(self, raw, expected):
         key = 'dummy-key'
         settings = {key: raw}
         _string_setting(settings, key, None)
         assert settings[key] == expected
 
     @pytest.mark.parametrize('raw, expected', [
         ('', []),
         ('test', ['test']),
         ('CaSe-TeSt', ['CaSe-TeSt']),
         ('test-string-烩€', ['test-string-烩€']),
         (u'test-string-烩€', [u'test-string-烩€']),
         ('hg git svn', ['hg', 'git', 'svn']),
         ('hg,git,svn', ['hg', 'git', 'svn']),
         ('hg, git, svn', ['hg', 'git', 'svn']),
         ('hg\ngit\nsvn', ['hg', 'git', 'svn']),
         (' hg\n git\n svn ', ['hg', 'git', 'svn']),
         (', hg , git , svn , ', ['', 'hg', 'git', 'svn', '']),
         ('cheese,free node,other', ['cheese', 'free node', 'other']),
-    ])
+    ], ids=no_newline_id_generator)
     def test_list_setting_helper(self, raw, expected):
         key = 'dummy-key'
         settings = {key: raw}
         _list_setting(settings, key, None)
         assert settings[key] == expected
 
     @pytest.mark.parametrize('raw, expected', [
         ('0', 0),
         ('-0', 0),
         ('12345', 12345),
         ('-12345', -12345),
         (u'-12345', -12345),
     ])
     def test_int_setting_helper(self, raw, expected):
         key = 'dummy-key'
         settings = {key: raw}
         _int_setting(settings, key, None)
         assert settings[key] == expected
 
     @pytest.mark.parametrize('raw', [
         ('0xff'),
         (''),
         ('invalid-int'),
         ('invalid-⁄~†'),
         (u'invalid-⁄~†'),
     ])
     def test_int_setting_helper_invalid_input(self, raw):
         key = 'dummy-key'
         settings = {key: raw}
         with pytest.raises(Exception):
             _int_setting(settings, key, None)
 
 
 class TestSanitizeVcsSettings(object):
     _bool_settings = [
         ('vcs.hooks.direct_calls', False),
         ('vcs.server.enable', True),
         ('vcs.start_server', False),
         ('startup.import_repos', False),
     ]
 
     _string_settings = [
         ('vcs.svn.compatible_version', ''),
         ('git_rev_filter', '--all'),
         ('vcs.hooks.protocol', 'http'),
         ('vcs.scm_app_implementation', 'http'),
         ('vcs.server', ''),
         ('vcs.server.log_level', 'debug'),
         ('vcs.server.protocol', 'http'),
     ]
 
     _list_settings = [
         ('vcs.backends', 'hg git'),
     ]
 
     @pytest.mark.parametrize('key, default', _list_settings)
     def test_list_setting_spacesep_list(self, key, default):
         test_list = ['test', 'list', 'values', 'for', key]
         input_value = ' '.join(test_list)
         settings = {key: input_value}
         _sanitize_vcs_settings(settings)
         assert settings[key] == test_list
 
     @pytest.mark.parametrize('key, default', _list_settings)
     def test_list_setting_newlinesep_list(self, key, default):
         test_list = ['test', 'list', 'values', 'for', key]
         input_value = '\n'.join(test_list)
         settings = {key: input_value}
         _sanitize_vcs_settings(settings)
         assert settings[key] == test_list
 
     @pytest.mark.parametrize('key, default', _list_settings)
     def test_list_setting_commasep_list(self, key, default):
         test_list = ['test', 'list', 'values', 'for', key]
         input_value = ','.join(test_list)
         settings = {key: input_value}
         _sanitize_vcs_settings(settings)
         assert settings[key] == test_list
 
     @pytest.mark.parametrize('key, default', _list_settings)
     def test_list_setting_comma_and_space_sep_list(self, key, default):
         test_list = ['test', 'list', 'values', 'for', key]
         input_value = ', '.join(test_list)
         settings = {key: input_value}
         _sanitize_vcs_settings(settings)
         assert settings[key] == test_list
 
     @pytest.mark.parametrize('key, default', _string_settings)
     def test_string_setting_string(self, key, default):
         test_value = 'test-string-for-{}'.format(key)
         settings = {key: test_value}
         _sanitize_vcs_settings(settings)
         assert settings[key] == test_value
 
     @pytest.mark.parametrize('key, default', _string_settings)
     def test_string_setting_default(self, key, default):
         settings = {}
         _sanitize_vcs_settings(settings)
         assert settings[key] == default
 
     @pytest.mark.parametrize('key, default', _string_settings)
     def test_string_setting_lowercase(self, key, default):
         test_value = 'Test-String-For-{}'.format(key)
         settings = {key: test_value}
         _sanitize_vcs_settings(settings)
         assert settings[key] == test_value.lower()
 
     @pytest.mark.parametrize('key, default', _bool_settings)
     def test_bool_setting_true(self, key, default):
         settings = {key: 'true'}
         _sanitize_vcs_settings(settings)
         assert settings[key] is True
 
     @pytest.mark.parametrize('key, default', _bool_settings)
     def test_bool_setting_false(self, key, default):
         settings = {key: 'false'}
         _sanitize_vcs_settings(settings)
         assert settings[key] is False
 
     @pytest.mark.parametrize('key, default', _bool_settings)
     def test_bool_setting_invalid_string(self, key, default):
         settings = {key: 'no-bool-val-string'}
         _sanitize_vcs_settings(settings)
         assert settings[key] is False
 
     @pytest.mark.parametrize('key, default', _bool_settings)
     def test_bool_setting_default(self, key, default):
         settings = {}
         _sanitize_vcs_settings(settings)
         assert settings[key] is default

rhodecode/tests/controllers/test_repo_groups.py

@@ -1,79 +1,79 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 pylons
 import pytest
 from pylons import tmpl_context as c
 
 
 @pytest.fixture
 def current_user(request, user_util):
     user = user_util.create_user()
 
     request_mock = mock.Mock(user=user)
     pylons.request._push_object(request_mock)
 
     @request.addfinalizer
     def cleanup():
         pylons.request._pop_object()
     return user
 
 
 @pytest.fixture
 def personal_group_with_parent(user_util, current_user):
     group_read_only = user_util.create_repo_group()
     user_util.grant_user_permission_to_repo_group(
         group_read_only, current_user, 'group.read')
 
     # TODO: johbo: This should go into the business models
     group_as_admin = user_util.create_repo_group(
         owner=current_user)
     group_as_admin.parent_group = group_read_only
     group_as_admin.group_name = group_as_admin.get_new_name(
         group_as_admin.group_name)
 
     return group_as_admin
 
 
 @pytest.fixture
 def controller():
     from rhodecode.controllers.admin import repo_groups
     return repo_groups.RepoGroupsController()
 
 
 def test_repo_groups_load_defaults(
         current_user, personal_group_with_parent, controller):
     personal_group = personal_group_with_parent
     controller._RepoGroupsController__load_defaults(True, personal_group)
 
-    expected_list = ['-1', personal_group.parent_group.group_id]
+    expected_list = [-1, personal_group.parent_group.group_id]
     returned_group_ids = [group[0] for group in c.repo_groups]
     assert returned_group_ids == expected_list
 
 
 def test_repo_groups_load_defaults_with_missing_group(
         current_user, personal_group_with_parent, controller):
     personal_group = personal_group_with_parent
     controller._RepoGroupsController__load_defaults(True)
 
-    expected_list = sorted(['-1', personal_group.group_id])
+    expected_list = sorted([-1, personal_group.group_id])
     returned_group_ids = sorted([group[0] for group in c.repo_groups])
     assert returned_group_ids == expected_list

rhodecode/tests/functional/test_admin_repos.py

@@ -1,1268 +1,1207 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 urllib
 
 import mock
 import pytest
 
 from rhodecode.lib import auth
 from rhodecode.lib.utils2 import safe_str, str2bool, safe_unicode
 from rhodecode.lib.vcs.exceptions import RepositoryRequirementError
 from rhodecode.model.db import Repository, RepoGroup, UserRepoToPerm, User,\
     Permission
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.settings import SettingsModel, VcsSettingsModel
 from rhodecode.model.user import UserModel
 from rhodecode.tests import (
     login_user_session, url, assert_session_flash, TEST_USER_ADMIN_LOGIN,
     TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS, HG_REPO, GIT_REPO,
     logout_user_session)
 from rhodecode.tests.fixture import Fixture, error_function
 from rhodecode.tests.utils import AssertResponse, repo_on_filesystem
 
 fixture = Fixture()
 
 
 @pytest.mark.usefixtures("app")
 class TestAdminRepos(object):
 
     def test_index(self):
         self.app.get(url('repos'))
 
     def test_create_page_restricted(self, autologin_user, backend):
         with mock.patch('rhodecode.BACKENDS', {'git': 'git'}):
             response = self.app.get(url('new_repo'), status=200)
         assert_response = AssertResponse(response)
         element = assert_response.get_element('#repo_type')
         assert element.text_content() == '\ngit\n'
 
     def test_create_page_non_restricted(self, autologin_user, backend):
         response = self.app.get(url('new_repo'), status=200)
         assert_response = AssertResponse(response)
         assert_response.element_contains('#repo_type', 'git')
         assert_response.element_contains('#repo_type', 'svn')
         assert_response.element_contains('#repo_type', 'hg')
 
     @pytest.mark.parametrize("suffix",
                              [u'', u'xxa'], ids=['', 'non-ascii'])
     def test_create(self, autologin_user, backend, suffix, csrf_token):
         repo_name_unicode = backend.new_repo_name(suffix=suffix)
         repo_name = repo_name_unicode.encode('utf8')
         description_unicode = u'description for newly created repo' + suffix
         description = description_unicode.encode('utf8')
         response = self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 csrf_token=csrf_token),
             status=302)
 
         self.assert_repository_is_created_correctly(
             repo_name, description, backend)
 
     def test_create_numeric(self, autologin_user, backend, csrf_token):
         numeric_repo = '1234'
         repo_name = numeric_repo
         description = 'description for newly created repo' + numeric_repo
         self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 csrf_token=csrf_token))
 
         self.assert_repository_is_created_correctly(
             repo_name, description, backend)
 
     @pytest.mark.parametrize("suffix", [u'', u'ąćę'], ids=['', 'non-ascii'])
     def test_create_in_group(
             self, autologin_user, backend, suffix, csrf_token):
         # create GROUP
         group_name = 'sometest_%s' % backend.alias
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
 
         repo_name = u'ingroup' + suffix
         repo_name_full = RepoGroup.url_sep().join(
             [group_name, repo_name])
         description = u'description for newly created repo'
         self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=safe_str(repo_name),
                 repo_type=backend.alias,
                 repo_description=description,
                 repo_group=gr.group_id,
                 csrf_token=csrf_token))
 
         # TODO: johbo: Cleanup work to fixture
         try:
             self.assert_repository_is_created_correctly(
                 repo_name_full, description, backend)
 
             new_repo = RepoModel().get_by_repo_name(repo_name_full)
             inherited_perms = UserRepoToPerm.query().filter(
                 UserRepoToPerm.repository_id == new_repo.repo_id).all()
             assert len(inherited_perms) == 1
         finally:
             RepoModel().delete(repo_name_full)
             RepoGroupModel().delete(group_name)
             Session().commit()
 
     def test_create_in_group_numeric(
             self, autologin_user, backend, csrf_token):
         # create GROUP
         group_name = 'sometest_%s' % backend.alias
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
 
         repo_name = '12345'
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = 'description for newly created repo'
         self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 repo_group=gr.group_id,
                 csrf_token=csrf_token))
 
         # TODO: johbo: Cleanup work to fixture
         try:
             self.assert_repository_is_created_correctly(
                 repo_name_full, description, backend)
 
             new_repo = RepoModel().get_by_repo_name(repo_name_full)
             inherited_perms = UserRepoToPerm.query()\
                 .filter(UserRepoToPerm.repository_id == new_repo.repo_id).all()
             assert len(inherited_perms) == 1
         finally:
             RepoModel().delete(repo_name_full)
             RepoGroupModel().delete(group_name)
             Session().commit()
 
     def test_create_in_group_without_needed_permissions(self, backend):
         session = login_user_session(
             self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         csrf_token = auth.get_csrf_token(session)
         # revoke
         user_model = UserModel()
         # disable fork and create on default user
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
 
         # disable on regular user
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
         Session().commit()
 
         # create GROUP
         group_name = 'reg_sometest_%s' % backend.alias
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
 
         group_name_allowed = 'reg_sometest_allowed_%s' % backend.alias
         gr_allowed = RepoGroupModel().create(
             group_name=group_name_allowed,
             group_description='test',
             owner=TEST_USER_REGULAR_LOGIN)
         Session().commit()
 
         repo_name = 'ingroup'
         description = 'description for newly created repo'
         response = self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 repo_group=gr.group_id,
                 csrf_token=csrf_token))
 
         response.mustcontain('Invalid value')
 
         # user is allowed to create in this group
         repo_name = 'ingroup'
         repo_name_full = RepoGroup.url_sep().join(
             [group_name_allowed, repo_name])
         description = 'description for newly created repo'
         response = self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 repo_group=gr_allowed.group_id,
                 csrf_token=csrf_token))
 
         # TODO: johbo: Cleanup in pytest fixture
         try:
             self.assert_repository_is_created_correctly(
                 repo_name_full, description, backend)
 
             new_repo = RepoModel().get_by_repo_name(repo_name_full)
             inherited_perms = UserRepoToPerm.query().filter(
                 UserRepoToPerm.repository_id == new_repo.repo_id).all()
             assert len(inherited_perms) == 1
 
             assert repo_on_filesystem(repo_name_full)
         finally:
             RepoModel().delete(repo_name_full)
             RepoGroupModel().delete(group_name)
             RepoGroupModel().delete(group_name_allowed)
             Session().commit()
 
     def test_create_in_group_inherit_permissions(self, autologin_user, backend,
                                                  csrf_token):
         # create GROUP
         group_name = 'sometest_%s' % backend.alias
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         perm = Permission.get_by_key('repository.write')
         RepoGroupModel().grant_user_permission(
             gr, TEST_USER_REGULAR_LOGIN, perm)
 
         # add repo permissions
         Session().commit()
 
         repo_name = 'ingroup_inherited_%s' % backend.alias
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = 'description for newly created repo'
         self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 repo_group=gr.group_id,
                 repo_copy_permissions=True,
                 csrf_token=csrf_token))
 
         # TODO: johbo: Cleanup to pytest fixture
         try:
             self.assert_repository_is_created_correctly(
                 repo_name_full, description, backend)
         except Exception:
             RepoGroupModel().delete(group_name)
             Session().commit()
             raise
 
         # check if inherited permissions are applied
         new_repo = RepoModel().get_by_repo_name(repo_name_full)
         inherited_perms = UserRepoToPerm.query().filter(
             UserRepoToPerm.repository_id == new_repo.repo_id).all()
         assert len(inherited_perms) == 2
 
         assert TEST_USER_REGULAR_LOGIN in [
             x.user.username for x in inherited_perms]
         assert 'repository.write' in [
             x.permission.permission_name for x in inherited_perms]
 
         RepoModel().delete(repo_name_full)
         RepoGroupModel().delete(group_name)
         Session().commit()
 
     @pytest.mark.xfail_backends(
         "git", "hg", reason="Missing reposerver support")
     def test_create_with_clone_uri(self, autologin_user, backend, reposerver,
                                    csrf_token):
         source_repo = backend.create_repo(number_of_commits=2)
         source_repo_name = source_repo.repo_name
         reposerver.serve(source_repo.scm_instance())
 
         repo_name = backend.new_repo_name()
         response = self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description='',
                 clone_uri=reposerver.url,
                 csrf_token=csrf_token),
             status=302)
 
         # Should be redirected to the creating page
         response.mustcontain('repo_creating')
 
         # Expecting that both repositories have same history
         source_repo = RepoModel().get_by_repo_name(source_repo_name)
         source_vcs = source_repo.scm_instance()
         repo = RepoModel().get_by_repo_name(repo_name)
         repo_vcs = repo.scm_instance()
         assert source_vcs[0].message == repo_vcs[0].message
         assert source_vcs.count() == repo_vcs.count()
         assert source_vcs.commit_ids == repo_vcs.commit_ids
 
     @pytest.mark.xfail_backends("svn", reason="Depends on import support")
     def test_create_remote_repo_wrong_clone_uri(self, autologin_user, backend,
                                                 csrf_token):
         repo_name = backend.new_repo_name()
         description = 'description for newly created repo'
         response = self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 clone_uri='http://repo.invalid/repo',
                 csrf_token=csrf_token))
         response.mustcontain('invalid clone url')
 
     @pytest.mark.xfail_backends("svn", reason="Depends on import support")
     def test_create_remote_repo_wrong_clone_uri_hg_svn(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.new_repo_name()
         description = 'description for newly created repo'
         response = self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 clone_uri='svn+http://svn.invalid/repo',
                 csrf_token=csrf_token))
         response.mustcontain('invalid clone url')
 
     def test_create_with_git_suffix(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.new_repo_name() + ".git"
         description = 'description for newly created repo'
         response = self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 csrf_token=csrf_token))
         response.mustcontain('Repository name cannot end with .git')
 
     @pytest.mark.parametrize("suffix", [u'', u'ąęł'], ids=['', 'non-ascii'])
     def test_delete(self, autologin_user, backend, suffix, csrf_token):
         repo = backend.create_repo(name_suffix=suffix)
         repo_name = repo.repo_name
 
         response = self.app.post(url('repo', repo_name=repo_name),
                                  params={'_method': 'delete',
                                          'csrf_token': csrf_token})
         assert_session_flash(response, 'Deleted repository %s' % (repo_name))
         response.follow()
 
         # check if repo was deleted from db
         assert RepoModel().get_by_repo_name(repo_name) is None
         assert not repo_on_filesystem(repo_name)
 
     def test_show(self, autologin_user, backend):
         self.app.get(url('repo', repo_name=backend.repo_name))
 
-    def test_edit(self, backend, autologin_user):
-        self.app.get(url('edit_repo', repo_name=backend.repo_name))
-
-    def test_edit_accessible_when_missing_requirements(
-            self, backend_hg, autologin_user):
-        scm_patcher = mock.patch.object(
-            Repository, 'scm_instance', side_effect=RepositoryRequirementError)
-        with scm_patcher:
-            self.app.get(url('edit_repo', repo_name=backend_hg.repo_name))
-
-    def test_set_private_flag_sets_default_to_none(
-            self, autologin_user, backend, csrf_token):
-        # initially repository perm should be read
-        perm = _get_permission_for_user(user='default', repo=backend.repo_name)
-        assert len(perm) == 1
-        assert perm[0].permission.permission_name == 'repository.read'
-        assert not backend.repo.private
-
-        response = self.app.post(
-            url('repo', repo_name=backend.repo_name),
-            fixture._get_repo_create_params(
-                repo_private=1,
-                repo_name=backend.repo_name,
-                repo_type=backend.alias,
-                user=TEST_USER_ADMIN_LOGIN,
-                _method='put',
-                csrf_token=csrf_token))
-        assert_session_flash(
-            response,
-            msg='Repository %s updated successfully' % (backend.repo_name))
-        assert backend.repo.private
-
-        # now the repo default permission should be None
-        perm = _get_permission_for_user(user='default', repo=backend.repo_name)
-        assert len(perm) == 1
-        assert perm[0].permission.permission_name == 'repository.none'
-
-        response = self.app.post(
-            url('repo', repo_name=backend.repo_name),
-            fixture._get_repo_create_params(
-                repo_private=False,
-                repo_name=backend.repo_name,
-                repo_type=backend.alias,
-                user=TEST_USER_ADMIN_LOGIN,
-                _method='put',
-                csrf_token=csrf_token))
-        assert_session_flash(
-            response,
-            msg='Repository %s updated successfully' % (backend.repo_name))
-        assert not backend.repo.private
-
-        # we turn off private now the repo default permission should stay None
-        perm = _get_permission_for_user(user='default', repo=backend.repo_name)
-        assert len(perm) == 1
-        assert perm[0].permission.permission_name == 'repository.none'
-
-        # update this permission back
-        perm[0].permission = Permission.get_by_key('repository.read')
-        Session().add(perm[0])
-        Session().commit()
-
     def test_default_user_cannot_access_private_repo_in_a_group(
             self, autologin_user, user_util, backend, csrf_token):
 
         group = user_util.create_repo_group()
 
         repo = backend.create_repo(
             repo_private=True, repo_group=group, repo_copy_permissions=True)
 
         permissions = _get_permission_for_user(
             user='default', repo=repo.repo_name)
         assert len(permissions) == 1
         assert permissions[0].permission.permission_name == 'repository.none'
         assert permissions[0].repository.private is True
 
     def test_set_repo_fork_has_no_self_id(self, autologin_user, backend):
         repo = backend.repo
         response = self.app.get(
             url('edit_repo_advanced', repo_name=backend.repo_name))
         opt = """<option value="%s">vcs_test_git</option>""" % repo.repo_id
         response.mustcontain(no=[opt])
 
     def test_set_fork_of_target_repo(
             self, autologin_user, backend, csrf_token):
         target_repo = 'target_%s' % backend.alias
         fixture.create_repo(target_repo, repo_type=backend.alias)
         repo2 = Repository.get_by_repo_name(target_repo)
         response = self.app.post(
             url('edit_repo_advanced_fork', repo_name=backend.repo_name),
             params={'id_fork_of': repo2.repo_id, '_method': 'put',
                     'csrf_token': csrf_token})
         repo = Repository.get_by_repo_name(backend.repo_name)
         repo2 = Repository.get_by_repo_name(target_repo)
         assert_session_flash(
             response,
             'Marked repo %s as fork of %s' % (repo.repo_name, repo2.repo_name))
 
         assert repo.fork == repo2
         response = response.follow()
         # check if given repo is selected
 
         opt = 'This repository is a fork of <a href="%s">%s</a>' % (
             url('summary_home', repo_name=repo2.repo_name), repo2.repo_name)
 
         response.mustcontain(opt)
 
         fixture.destroy_repo(target_repo, forks='detach')
 
     @pytest.mark.backends("hg", "git")
     def test_set_fork_of_other_type_repo(self, autologin_user, backend,
                                          csrf_token):
         TARGET_REPO_MAP = {
             'git': {
                 'type': 'hg',
                 'repo_name': HG_REPO},
             'hg': {
                 'type': 'git',
                 'repo_name': GIT_REPO},
         }
         target_repo = TARGET_REPO_MAP[backend.alias]
 
         repo2 = Repository.get_by_repo_name(target_repo['repo_name'])
         response = self.app.post(
             url('edit_repo_advanced_fork', repo_name=backend.repo_name),
             params={'id_fork_of': repo2.repo_id, '_method': 'put',
                     'csrf_token': csrf_token})
         assert_session_flash(
             response,
             'Cannot set repository as fork of repository with other type')
 
     def test_set_fork_of_none(self, autologin_user, backend, csrf_token):
         # mark it as None
         response = self.app.post(
             url('edit_repo_advanced_fork', repo_name=backend.repo_name),
             params={'id_fork_of': None, '_method': 'put',
                     'csrf_token': csrf_token})
         assert_session_flash(
             response,
             'Marked repo %s as fork of %s'
             % (backend.repo_name, "Nothing"))
         assert backend.repo.fork is None
 
     def test_set_fork_of_same_repo(self, autologin_user, backend, csrf_token):
         repo = Repository.get_by_repo_name(backend.repo_name)
         response = self.app.post(
             url('edit_repo_advanced_fork', repo_name=backend.repo_name),
             params={'id_fork_of': repo.repo_id, '_method': 'put',
                     'csrf_token': csrf_token})
         assert_session_flash(
             response, 'An error occurred during this operation')
 
     def test_create_on_top_level_without_permissions(self, backend):
         session = login_user_session(
             self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         csrf_token = auth.get_csrf_token(session)
 
         # revoke
         user_model = UserModel()
         # disable fork and create on default user
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
 
         # disable on regular user
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
         Session().commit()
 
         repo_name = backend.new_repo_name()
         description = 'description for newly created repo'
         response = self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 csrf_token=csrf_token))
 
         response.mustcontain(
             u"You do not have the permission to store repositories in "
             u"the root location.")
 
     @mock.patch.object(RepoModel, '_create_filesystem_repo', error_function)
     def test_create_repo_when_filesystem_op_fails(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.new_repo_name()
         description = 'description for newly created repo'
 
         response = self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 csrf_token=csrf_token))
 
         assert_session_flash(
             response, 'Error creating repository %s' % repo_name)
         # repo must not be in db
         assert backend.repo is None
         # repo must not be in filesystem !
         assert not repo_on_filesystem(repo_name)
 
     def assert_repository_is_created_correctly(
             self, repo_name, description, backend):
         repo_name_utf8 = safe_str(repo_name)
 
         # run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name))
         assert response.json == {u'result': True}
 
         flash_msg = u'Created repository <a href="/{}">{}</a>'.format(
             urllib.quote(repo_name_utf8), repo_name)
         assert_session_flash(response, flash_msg)
 
         # test if the repo was created in the database
         new_repo = RepoModel().get_by_repo_name(repo_name)
 
         assert new_repo.repo_name == repo_name
         assert new_repo.description == description
 
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(backend.alias)
 
         assert repo_on_filesystem(repo_name)
 
 
 @pytest.mark.usefixtures("app")
 class TestVcsSettings(object):
     FORM_DATA = {
         'inherit_global_settings': False,
         'hooks_changegroup_repo_size': False,
         'hooks_changegroup_push_logger': False,
         'hooks_outgoing_pull_logger': False,
         'extensions_largefiles': False,
         'phases_publish': 'False',
         'rhodecode_pr_merge_enabled': False,
         'rhodecode_use_outdated_comments': False,
         'new_svn_branch': '',
         'new_svn_tag': ''
     }
 
     @pytest.mark.skip_backends('svn')
     def test_global_settings_initial_values(self, autologin_user, backend):
         repo_name = backend.repo_name
         response = self.app.get(url('repo_vcs_settings', repo_name=repo_name))
 
         expected_settings = (
             'rhodecode_use_outdated_comments', 'rhodecode_pr_merge_enabled',
             'hooks_changegroup_repo_size', 'hooks_changegroup_push_logger',
             'hooks_outgoing_pull_logger'
         )
         for setting in expected_settings:
             self.assert_repo_value_equals_global_value(response, setting)
 
     def test_show_settings_requires_repo_admin_permission(
             self, backend, user_util, settings_util):
         repo = backend.create_repo()
         repo_name = repo.repo_name
         user = UserModel().get_by_username(TEST_USER_REGULAR_LOGIN)
         user_util.grant_user_permission_to_repo(repo, user, 'repository.admin')
         login_user_session(
             self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         self.app.get(url('repo_vcs_settings', repo_name=repo_name), status=200)
 
     def test_inherit_global_settings_flag_is_true_by_default(
             self, autologin_user, backend):
         repo_name = backend.repo_name
         response = self.app.get(url('repo_vcs_settings', repo_name=repo_name))
 
         assert_response = AssertResponse(response)
         element = assert_response.get_element('#inherit_global_settings')
         assert element.checked
 
     @pytest.mark.parametrize('checked_value', [True, False])
     def test_inherit_global_settings_value(
             self, autologin_user, backend, checked_value, settings_util):
         repo = backend.create_repo()
         repo_name = repo.repo_name
         settings_util.create_repo_rhodecode_setting(
             repo, 'inherit_vcs_settings', checked_value, 'bool')
         response = self.app.get(url('repo_vcs_settings', repo_name=repo_name))
 
         assert_response = AssertResponse(response)
         element = assert_response.get_element('#inherit_global_settings')
         assert element.checked == checked_value
 
     @pytest.mark.skip_backends('svn')
     def test_hooks_settings_are_created(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             for section, key in VcsSettingsModel.HOOKS_SETTINGS:
                 ui = settings.get_ui_by_section_and_key(section, key)
                 assert ui.ui_active is False
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_hooks_settings_are_not_created_for_svn(
             self, autologin_user, backend_svn, csrf_token):
         repo_name = backend_svn.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             for section, key in VcsSettingsModel.HOOKS_SETTINGS:
                 ui = settings.get_ui_by_section_and_key(section, key)
                 assert ui is None
         finally:
             self._cleanup_repo_settings(settings)
 
     @pytest.mark.skip_backends('svn')
     def test_hooks_settings_are_updated(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         settings = SettingsModel(repo=repo_name)
         for section, key in VcsSettingsModel.HOOKS_SETTINGS:
             settings.create_ui_section_value(section, '', key=key, active=True)
 
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         try:
             for section, key in VcsSettingsModel.HOOKS_SETTINGS:
                 ui = settings.get_ui_by_section_and_key(section, key)
                 assert ui.ui_active is False
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_hooks_settings_are_not_updated_for_svn(
             self, autologin_user, backend_svn, csrf_token):
         repo_name = backend_svn.repo_name
         settings = SettingsModel(repo=repo_name)
         for section, key in VcsSettingsModel.HOOKS_SETTINGS:
             settings.create_ui_section_value(section, '', key=key, active=True)
 
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         try:
             for section, key in VcsSettingsModel.HOOKS_SETTINGS:
                 ui = settings.get_ui_by_section_and_key(section, key)
                 assert ui.ui_active is True
         finally:
             self._cleanup_repo_settings(settings)
 
     @pytest.mark.skip_backends('svn')
     def test_pr_settings_are_created(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             for name in VcsSettingsModel.GENERAL_SETTINGS:
                 setting = settings.get_setting_by_name(name)
                 assert setting.app_settings_value is False
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_pr_settings_are_not_created_for_svn(
             self, autologin_user, backend_svn, csrf_token):
         repo_name = backend_svn.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             for name in VcsSettingsModel.GENERAL_SETTINGS:
                 setting = settings.get_setting_by_name(name)
                 assert setting is None
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_pr_settings_creation_requires_repo_admin_permission(
             self, backend, user_util, settings_util, csrf_token):
         repo = backend.create_repo()
         repo_name = repo.repo_name
 
         logout_user_session(self.app, csrf_token)
         session = login_user_session(
             self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         new_csrf_token = auth.get_csrf_token(session)
 
         user = UserModel().get_by_username(TEST_USER_REGULAR_LOGIN)
         repo = Repository.get_by_repo_name(repo_name)
         user_util.grant_user_permission_to_repo(repo, user, 'repository.admin')
         data = self.FORM_DATA.copy()
         data['csrf_token'] = new_csrf_token
         settings = SettingsModel(repo=repo_name)
 
         try:
             self.app.post(
                 url('repo_vcs_settings', repo_name=repo_name), data,
                 status=302)
         finally:
             self._cleanup_repo_settings(settings)
 
     @pytest.mark.skip_backends('svn')
     def test_pr_settings_are_updated(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         settings = SettingsModel(repo=repo_name)
         for name in VcsSettingsModel.GENERAL_SETTINGS:
             settings.create_or_update_setting(name, True, 'bool')
 
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         try:
             for name in VcsSettingsModel.GENERAL_SETTINGS:
                 setting = settings.get_setting_by_name(name)
                 assert setting.app_settings_value is False
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_pr_settings_are_not_updated_for_svn(
             self, autologin_user, backend_svn, csrf_token):
         repo_name = backend_svn.repo_name
         settings = SettingsModel(repo=repo_name)
         for name in VcsSettingsModel.GENERAL_SETTINGS:
             settings.create_or_update_setting(name, True, 'bool')
 
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         try:
             for name in VcsSettingsModel.GENERAL_SETTINGS:
                 setting = settings.get_setting_by_name(name)
                 assert setting.app_settings_value is True
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_svn_settings_are_created(
             self, autologin_user, backend_svn, csrf_token, settings_util):
         repo_name = backend_svn.repo_name
         data = self.FORM_DATA.copy()
         data['new_svn_tag'] = 'svn-tag'
         data['new_svn_branch'] = 'svn-branch'
         data['csrf_token'] = csrf_token
 
         # Create few global settings to make sure that uniqueness validators
         # are not triggered
         settings_util.create_rhodecode_ui(
             VcsSettingsModel.SVN_BRANCH_SECTION, 'svn-branch')
         settings_util.create_rhodecode_ui(
             VcsSettingsModel.SVN_TAG_SECTION, 'svn-tag')
 
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             svn_branches = settings.get_ui_by_section(
                 VcsSettingsModel.SVN_BRANCH_SECTION)
             svn_branch_names = [b.ui_value for b in svn_branches]
             svn_tags = settings.get_ui_by_section(
                 VcsSettingsModel.SVN_TAG_SECTION)
             svn_tag_names = [b.ui_value for b in svn_tags]
             assert 'svn-branch' in svn_branch_names
             assert 'svn-tag' in svn_tag_names
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_svn_settings_are_unique(
             self, autologin_user, backend_svn, csrf_token, settings_util):
         repo = backend_svn.repo
         repo_name = repo.repo_name
         data = self.FORM_DATA.copy()
         data['new_svn_tag'] = 'test_tag'
         data['new_svn_branch'] = 'test_branch'
         data['csrf_token'] = csrf_token
         settings_util.create_repo_rhodecode_ui(
             repo, VcsSettingsModel.SVN_BRANCH_SECTION, 'test_branch')
         settings_util.create_repo_rhodecode_ui(
             repo, VcsSettingsModel.SVN_TAG_SECTION, 'test_tag')
 
         response = self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=200)
         response.mustcontain('Pattern already exists')
 
     def test_svn_settings_with_empty_values_are_not_created(
             self, autologin_user, backend_svn, csrf_token):
         repo_name = backend_svn.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             svn_branches = settings.get_ui_by_section(
                 VcsSettingsModel.SVN_BRANCH_SECTION)
             svn_tags = settings.get_ui_by_section(
                 VcsSettingsModel.SVN_TAG_SECTION)
             assert len(svn_branches) == 0
             assert len(svn_tags) == 0
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_svn_settings_are_shown_for_svn_repository(
             self, autologin_user, backend_svn, csrf_token):
         repo_name = backend_svn.repo_name
         response = self.app.get(
             url('repo_vcs_settings', repo_name=repo_name), status=200)
         response.mustcontain('Subversion Settings')
 
     @pytest.mark.skip_backends('svn')
     def test_svn_settings_are_not_created_for_not_svn_repository(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             svn_branches = settings.get_ui_by_section(
                 VcsSettingsModel.SVN_BRANCH_SECTION)
             svn_tags = settings.get_ui_by_section(
                 VcsSettingsModel.SVN_TAG_SECTION)
             assert len(svn_branches) == 0
             assert len(svn_tags) == 0
         finally:
             self._cleanup_repo_settings(settings)
 
     @pytest.mark.skip_backends('svn')
     def test_svn_settings_are_shown_only_for_svn_repository(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         response = self.app.get(
             url('repo_vcs_settings', repo_name=repo_name), status=200)
         response.mustcontain(no='Subversion Settings')
 
     def test_hg_settings_are_created(
             self, autologin_user, backend_hg, csrf_token):
         repo_name = backend_hg.repo_name
         data = self.FORM_DATA.copy()
         data['new_svn_tag'] = 'svn-tag'
         data['new_svn_branch'] = 'svn-branch'
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             largefiles_ui = settings.get_ui_by_section_and_key(
                 'extensions', 'largefiles')
             assert largefiles_ui.ui_active is False
             phases_ui = settings.get_ui_by_section_and_key(
                 'phases', 'publish')
             assert str2bool(phases_ui.ui_value) is False
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_hg_settings_are_updated(
             self, autologin_user, backend_hg, csrf_token):
         repo_name = backend_hg.repo_name
         settings = SettingsModel(repo=repo_name)
         settings.create_ui_section_value(
             'extensions', '', key='largefiles', active=True)
         settings.create_ui_section_value(
             'phases', '1', key='publish', active=True)
 
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         try:
             largefiles_ui = settings.get_ui_by_section_and_key(
                 'extensions', 'largefiles')
             assert largefiles_ui.ui_active is False
             phases_ui = settings.get_ui_by_section_and_key(
                 'phases', 'publish')
             assert str2bool(phases_ui.ui_value) is False
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_hg_settings_are_shown_for_hg_repository(
             self, autologin_user, backend_hg, csrf_token):
         repo_name = backend_hg.repo_name
         response = self.app.get(
             url('repo_vcs_settings', repo_name=repo_name), status=200)
         response.mustcontain('Mercurial Settings')
 
     @pytest.mark.skip_backends('hg')
     def test_hg_settings_are_created_only_for_hg_repository(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             largefiles_ui = settings.get_ui_by_section_and_key(
                 'extensions', 'largefiles')
             assert largefiles_ui is None
             phases_ui = settings.get_ui_by_section_and_key(
                 'phases', 'publish')
             assert phases_ui is None
         finally:
             self._cleanup_repo_settings(settings)
 
     @pytest.mark.skip_backends('hg')
     def test_hg_settings_are_shown_only_for_hg_repository(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         response = self.app.get(
             url('repo_vcs_settings', repo_name=repo_name), status=200)
         response.mustcontain(no='Mercurial Settings')
 
     @pytest.mark.skip_backends('hg')
     def test_hg_settings_are_updated_only_for_hg_repository(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         settings = SettingsModel(repo=repo_name)
         settings.create_ui_section_value(
             'extensions', '', key='largefiles', active=True)
         settings.create_ui_section_value(
             'phases', '1', key='publish', active=True)
 
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         try:
             largefiles_ui = settings.get_ui_by_section_and_key(
                 'extensions', 'largefiles')
             assert largefiles_ui.ui_active is True
             phases_ui = settings.get_ui_by_section_and_key(
                 'phases', 'publish')
             assert phases_ui.ui_value == '1'
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_per_repo_svn_settings_are_displayed(
             self, autologin_user, backend_svn, settings_util):
         repo = backend_svn.create_repo()
         repo_name = repo.repo_name
         branches = [
             settings_util.create_repo_rhodecode_ui(
                 repo, VcsSettingsModel.SVN_BRANCH_SECTION,
                 'branch_{}'.format(i))
             for i in range(10)]
         tags = [
             settings_util.create_repo_rhodecode_ui(
                 repo, VcsSettingsModel.SVN_TAG_SECTION, 'tag_{}'.format(i))
             for i in range(10)]
 
         response = self.app.get(
             url('repo_vcs_settings', repo_name=repo_name), status=200)
         assert_response = AssertResponse(response)
         for branch in branches:
             css_selector = '[name=branch_value_{}]'.format(branch.ui_id)
             element = assert_response.get_element(css_selector)
             assert element.value == branch.ui_value
         for tag in tags:
             css_selector = '[name=tag_ui_value_new_{}]'.format(tag.ui_id)
             element = assert_response.get_element(css_selector)
             assert element.value == tag.ui_value
 
     def test_per_repo_hg_and_pr_settings_are_not_displayed_for_svn(
             self, autologin_user, backend_svn, settings_util):
         repo = backend_svn.create_repo()
         repo_name = repo.repo_name
         response = self.app.get(
             url('repo_vcs_settings', repo_name=repo_name), status=200)
         response.mustcontain(no='<label>Hooks:</label>')
         response.mustcontain(no='<label>Pull Request Settings:</label>')
 
     def test_inherit_global_settings_value_is_saved(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         data['inherit_global_settings'] = True
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
 
         settings = SettingsModel(repo=repo_name)
         vcs_settings = VcsSettingsModel(repo=repo_name)
         try:
             assert vcs_settings.inherit_global_settings is True
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_repo_cache_is_invalidated_when_settings_are_updated(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         data['inherit_global_settings'] = True
         settings = SettingsModel(repo=repo_name)
 
         invalidation_patcher = mock.patch(
             'rhodecode.controllers.admin.repos.ScmModel.mark_for_invalidation')
         with invalidation_patcher as invalidation_mock:
             self.app.post(
                 url('repo_vcs_settings', repo_name=repo_name), data,
                 status=302)
         try:
             invalidation_mock.assert_called_once_with(repo_name, delete=True)
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_other_settings_not_saved_inherit_global_settings_is_true(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         data['inherit_global_settings'] = True
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
 
         settings = SettingsModel(repo=repo_name)
         ui_settings = (
             VcsSettingsModel.HOOKS_SETTINGS + VcsSettingsModel.HG_SETTINGS)
 
         vcs_settings = []
         try:
             for section, key in ui_settings:
                 ui = settings.get_ui_by_section_and_key(section, key)
                 if ui:
                     vcs_settings.append(ui)
             vcs_settings.extend(settings.get_ui_by_section(
                 VcsSettingsModel.SVN_BRANCH_SECTION))
             vcs_settings.extend(settings.get_ui_by_section(
                 VcsSettingsModel.SVN_TAG_SECTION))
             for name in VcsSettingsModel.GENERAL_SETTINGS:
                 setting = settings.get_setting_by_name(name)
                 if setting:
                     vcs_settings.append(setting)
             assert vcs_settings == []
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_delete_svn_branch_and_tag_patterns(
             self, autologin_user, backend_svn, settings_util, csrf_token):
         repo = backend_svn.create_repo()
         repo_name = repo.repo_name
         branch = settings_util.create_repo_rhodecode_ui(
             repo, VcsSettingsModel.SVN_BRANCH_SECTION, 'test_branch',
             cleanup=False)
         tag = settings_util.create_repo_rhodecode_ui(
             repo, VcsSettingsModel.SVN_TAG_SECTION, 'test_tag', cleanup=False)
         data = {
             '_method': 'delete',
             'csrf_token': csrf_token
         }
         for id_ in (branch.ui_id, tag.ui_id):
             data['delete_svn_pattern'] = id_,
             self.app.post(
                 url('repo_vcs_settings', repo_name=repo_name), data,
                 headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=200)
         settings = VcsSettingsModel(repo=repo_name)
         assert settings.get_repo_svn_branch_patterns() == []
 
     def test_delete_svn_branch_requires_repo_admin_permission(
             self, backend_svn, user_util, settings_util, csrf_token):
         repo = backend_svn.create_repo()
         repo_name = repo.repo_name
 
         logout_user_session(self.app, csrf_token)
         session = login_user_session(
             self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         csrf_token = auth.get_csrf_token(session)
 
         repo = Repository.get_by_repo_name(repo_name)
         user = UserModel().get_by_username(TEST_USER_REGULAR_LOGIN)
         user_util.grant_user_permission_to_repo(repo, user, 'repository.admin')
         branch = settings_util.create_repo_rhodecode_ui(
             repo, VcsSettingsModel.SVN_BRANCH_SECTION, 'test_branch',
             cleanup=False)
         data = {
             '_method': 'delete',
             'csrf_token': csrf_token,
             'delete_svn_pattern': branch.ui_id
         }
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data,
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=200)
 
     def test_delete_svn_branch_raises_400_when_not_found(
             self, autologin_user, backend_svn, settings_util, csrf_token):
         repo_name = backend_svn.repo_name
         data = {
             '_method': 'delete',
             'delete_svn_pattern': 123,
             'csrf_token': csrf_token
         }
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data,
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=400)
 
     def test_delete_svn_branch_raises_400_when_no_id_specified(
             self, autologin_user, backend_svn, settings_util, csrf_token):
         repo_name = backend_svn.repo_name
         data = {
             '_method': 'delete',
             'csrf_token': csrf_token
         }
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data,
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=400)
 
     def _cleanup_repo_settings(self, settings_model):
         cleanup = []
         ui_settings = (
             VcsSettingsModel.HOOKS_SETTINGS + VcsSettingsModel.HG_SETTINGS)
 
         for section, key in ui_settings:
             ui = settings_model.get_ui_by_section_and_key(section, key)
             if ui:
                 cleanup.append(ui)
 
         cleanup.extend(settings_model.get_ui_by_section(
             VcsSettingsModel.INHERIT_SETTINGS))
         cleanup.extend(settings_model.get_ui_by_section(
             VcsSettingsModel.SVN_BRANCH_SECTION))
         cleanup.extend(settings_model.get_ui_by_section(
             VcsSettingsModel.SVN_TAG_SECTION))
 
         for name in VcsSettingsModel.GENERAL_SETTINGS:
             setting = settings_model.get_setting_by_name(name)
             if setting:
                 cleanup.append(setting)
 
         for object_ in cleanup:
             Session().delete(object_)
         Session().commit()
 
     def assert_repo_value_equals_global_value(self, response, setting):
         assert_response = AssertResponse(response)
         global_css_selector = '[name={}_inherited]'.format(setting)
         repo_css_selector = '[name={}]'.format(setting)
         repo_element = assert_response.get_element(repo_css_selector)
         global_element = assert_response.get_element(global_css_selector)
         assert repo_element.value == global_element.value
 
 
 def _get_permission_for_user(user, repo):
     perm = UserRepoToPerm.query()\
         .filter(UserRepoToPerm.repository ==
                 Repository.get_by_repo_name(repo))\
         .filter(UserRepoToPerm.user == User.get_by_username(user))\
         .all()
     return perm

rhodecode/tests/models/schemas/test_repo_group_schema.py

@@ -1,116 +1,116 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-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 colander
 import pytest
 
 from rhodecode.model.validation_schema import types
 from rhodecode.model.validation_schema.schemas import repo_group_schema
 
 
 class TestRepoGroupSchema(object):
 
     @pytest.mark.parametrize('given, expected', [
         ('my repo', 'my-repo'),
         (' hello   world mike ', 'hello-world-mike'),
 
         ('//group1/group2//', 'group1/group2'),
         ('//group1///group2//', 'group1/group2'),
         ('///group1/group2///group3', 'group1/group2/group3'),
         ('word g1/group2///group3', 'word-g1/group2/group3'),
 
         ('grou p1/gro;,,##up2//.../group3', 'grou-p1/group2/group3'),
 
         ('group,,,/,,,/1/2/3', 'group/1/2/3'),
         ('grou[]p1/gro;up2///gro up3', 'group1/group2/gro-up3'),
         (u'grou[]p1/gro;up2///gro up3/ąć', u'group1/group2/gro-up3/ąć'),
     ])
     def test_deserialize_repo_name(self, app, user_admin, given, expected):
         schema = repo_group_schema.RepoGroupSchema().bind()
         assert schema.get('repo_group_name').deserialize(given) == expected
 
     def test_deserialize(self, app, user_admin):
         schema = repo_group_schema.RepoGroupSchema().bind(
             user=user_admin
         )
 
         schema_data = schema.deserialize(dict(
-            repo_group_name='dupa',
+            repo_group_name='my_schema_group',
             repo_group_owner=user_admin.username
         ))
 
-        assert schema_data['repo_group_name'] == 'dupa'
+        assert schema_data['repo_group_name'] == u'my_schema_group'
         assert schema_data['repo_group'] == {
             'repo_group_id': None,
             'repo_group_name': types.RootLocation,
-            'repo_group_name_without_group': 'dupa'}
+            'repo_group_name_without_group': u'my_schema_group'}
 
     @pytest.mark.parametrize('given, err_key, expected_exc', [
-        ('xxx/dupa', 'repo_group', 'Parent repository group `xxx` does not exist'),
+        ('xxx/my_schema_group', 'repo_group', 'Parent repository group `xxx` does not exist'),
         ('', 'repo_group_name', 'Name must start with a letter or number. Got ``'),
     ])
     def test_deserialize_with_bad_group_name(
             self, app, user_admin, given, err_key, expected_exc):
         schema = repo_group_schema.RepoGroupSchema().bind(
             repo_type_options=['hg'],
             user=user_admin
         )
 
         with pytest.raises(colander.Invalid) as excinfo:
             schema.deserialize(dict(
                 repo_group_name=given,
                 repo_group_owner=user_admin.username
             ))
 
         assert excinfo.value.asdict()[err_key] == expected_exc
 
     def test_deserialize_with_group_name(self, app, user_admin, test_repo_group):
         schema = repo_group_schema.RepoGroupSchema().bind(
             user=user_admin
         )
 
-        full_name = test_repo_group.group_name + '/dupa'
+        full_name = test_repo_group.group_name + u'/my_schema_group'
         schema_data = schema.deserialize(dict(
             repo_group_name=full_name,
             repo_group_owner=user_admin.username
         ))
 
         assert schema_data['repo_group_name'] == full_name
         assert schema_data['repo_group'] == {
             'repo_group_id': test_repo_group.group_id,
             'repo_group_name': test_repo_group.group_name,
-            'repo_group_name_without_group': 'dupa'}
+            'repo_group_name_without_group': u'my_schema_group'}
 
     def test_deserialize_with_group_name_regular_user_no_perms(
             self, app, user_regular, test_repo_group):
         schema = repo_group_schema.RepoGroupSchema().bind(
             user=user_regular
         )
 
-        full_name = test_repo_group.group_name + '/dupa'
+        full_name = test_repo_group.group_name + u'/my_schema_group'
         with pytest.raises(colander.Invalid) as excinfo:
             schema.deserialize(dict(
                 repo_group_name=full_name,
                 repo_group_owner=user_regular.username
             ))
 
         expected = 'Parent repository group `{}` does not exist'.format(
             test_repo_group.group_name)
         assert excinfo.value.asdict()['repo_group'] == expected

rhodecode/tests/models/schemas/test_repo_schema.py

@@ -1,128 +1,130 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-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 colander
 import pytest
 
 from rhodecode.model.validation_schema import types
 from rhodecode.model.validation_schema.schemas import repo_schema
 
 
 class TestRepoSchema(object):
 
     #TODO:
     # test nested groups
 
     @pytest.mark.parametrize('given, expected', [
         ('my repo', 'my-repo'),
         (' hello   world mike ', 'hello-world-mike'),
 
         ('//group1/group2//', 'group1/group2'),
         ('//group1///group2//', 'group1/group2'),
         ('///group1/group2///group3', 'group1/group2/group3'),
         ('word g1/group2///group3', 'word-g1/group2/group3'),
 
         ('grou p1/gro;,,##up2//.../group3', 'grou-p1/group2/group3'),
 
         ('group,,,/,,,/1/2/3', 'group/1/2/3'),
         ('grou[]p1/gro;up2///gro up3', 'group1/group2/gro-up3'),
         (u'grou[]p1/gro;up2///gro up3/ąć', u'group1/group2/gro-up3/ąć'),
     ])
     def test_deserialize_repo_name(self, app, user_admin, given, expected):
 
         schema = repo_schema.RepoSchema().bind()
         assert expected == schema.get('repo_name').deserialize(given)
 
     def test_deserialize(self, app, user_admin):
         schema = repo_schema.RepoSchema().bind(
             repo_type_options=['hg'],
             user=user_admin
         )
 
         schema_data = schema.deserialize(dict(
-            repo_name='dupa',
+            repo_name='my_schema_repo',
             repo_type='hg',
             repo_owner=user_admin.username
         ))
 
-        assert schema_data['repo_name'] == 'dupa'
+        assert schema_data['repo_name'] == u'my_schema_repo'
         assert schema_data['repo_group'] == {
             'repo_group_id': None,
             'repo_group_name': types.RootLocation,
-            'repo_name_without_group': 'dupa'}
+            'repo_name_with_group': u'my_schema_repo',
+            'repo_name_without_group': u'my_schema_repo'}
 
     @pytest.mark.parametrize('given, err_key, expected_exc', [
-        ('xxx/dupa','repo_group', 'Repository group `xxx` does not exist'),
+        ('xxx/my_schema_repo','repo_group', 'Repository group `xxx` does not exist'),
         ('', 'repo_name', 'Name must start with a letter or number. Got ``'),
     ])
     def test_deserialize_with_bad_group_name(
             self, app, user_admin, given, err_key, expected_exc):
 
         schema = repo_schema.RepoSchema().bind(
             repo_type_options=['hg'],
             user=user_admin
         )
 
         with pytest.raises(colander.Invalid) as excinfo:
             schema.deserialize(dict(
                 repo_name=given,
                 repo_type='hg',
                 repo_owner=user_admin.username
             ))
 
         assert excinfo.value.asdict()[err_key] == expected_exc
 
     def test_deserialize_with_group_name(self, app, user_admin, test_repo_group):
         schema = repo_schema.RepoSchema().bind(
             repo_type_options=['hg'],
             user=user_admin
         )
 
-        full_name = test_repo_group.group_name + '/dupa'
+        full_name = test_repo_group.group_name + u'/my_schema_repo'
         schema_data = schema.deserialize(dict(
             repo_name=full_name,
             repo_type='hg',
             repo_owner=user_admin.username
         ))
 
         assert schema_data['repo_name'] == full_name
         assert schema_data['repo_group'] == {
             'repo_group_id': test_repo_group.group_id,
             'repo_group_name': test_repo_group.group_name,
-            'repo_name_without_group': 'dupa'}
+            'repo_name_with_group': full_name,
+            'repo_name_without_group': u'my_schema_repo'}
 
     def test_deserialize_with_group_name_regular_user_no_perms(
             self, app, user_regular, test_repo_group):
         schema = repo_schema.RepoSchema().bind(
             repo_type_options=['hg'],
             user=user_regular
         )
 
-        full_name = test_repo_group.group_name + '/dupa'
+        full_name = test_repo_group.group_name + '/my_schema_repo'
         with pytest.raises(colander.Invalid) as excinfo:
             schema.deserialize(dict(
                 repo_name=full_name,
                 repo_type='hg',
                 repo_owner=user_regular.username
             ))
 
         expected = 'Repository group `{}` does not exist'.format(
             test_repo_group.group_name)
         assert excinfo.value.asdict()['repo_group'] == expected

rhodecode/tests/models/test_scm.py

@@ -1,335 +1,336 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 stat
 import sys
 
 import pytest
 from mock import Mock, patch, DEFAULT
 
 import rhodecode
 from rhodecode.model import db, scm
+from rhodecode.tests import no_newline_id_generator
 
 
 def test_scm_instance_config(backend):
     repo = backend.create_repo()
     with patch.multiple('rhodecode.model.db.Repository',
                         _get_instance=DEFAULT,
                         _get_instance_cached=DEFAULT) as mocks:
         repo.scm_instance()
         mocks['_get_instance'].assert_called_with(
             config=None, cache=False)
 
         config = {'some': 'value'}
         repo.scm_instance(config=config)
         mocks['_get_instance'].assert_called_with(
             config=config, cache=False)
 
         with patch.dict(rhodecode.CONFIG, {'vcs_full_cache': 'true'}):
             repo.scm_instance(config=config)
             mocks['_get_instance_cached'].assert_called()
 
 
 def test__get_instance_config(backend):
     repo = backend.create_repo()
     vcs_class = Mock()
     with patch.multiple('rhodecode.lib.vcs.backends',
                         get_scm=DEFAULT,
                         get_backend=DEFAULT) as mocks:
         mocks['get_scm'].return_value = backend.alias
         mocks['get_backend'].return_value = vcs_class
         with patch('rhodecode.model.db.Repository._config') as config_mock:
             repo._get_instance()
             vcs_class.assert_called_with(
                 repo_path=repo.repo_full_path, config=config_mock,
                 create=False, with_wire={'cache': True})
 
         new_config = {'override': 'old_config'}
         repo._get_instance(config=new_config)
         vcs_class.assert_called_with(
             repo_path=repo.repo_full_path, config=new_config, create=False,
             with_wire={'cache': True})
 
 
 def test_mark_for_invalidation_config(backend):
     repo = backend.create_repo()
     with patch('rhodecode.model.db.Repository.update_commit_cache') as _mock:
         scm.ScmModel().mark_for_invalidation(repo.repo_name)
         _, kwargs = _mock.call_args
         assert kwargs['config'].__dict__ == repo._config.__dict__
 
 
 def test_mark_for_invalidation_with_delete_updates_last_commit(backend):
     commits = [{'message': 'A'}, {'message': 'B'}]
     repo = backend.create_repo(commits=commits)
     scm.ScmModel().mark_for_invalidation(repo.repo_name, delete=True)
     assert repo.changeset_cache['revision'] == 1
 
 
 def test_mark_for_invalidation_with_delete_updates_last_commit_empty(backend):
     repo = backend.create_repo()
     scm.ScmModel().mark_for_invalidation(repo.repo_name, delete=True)
     assert repo.changeset_cache['revision'] == -1
 
 
 def test_strip_with_multiple_heads(backend_hg):
     commits = [
         {'message': 'A'},
         {'message': 'a'},
         {'message': 'b'},
         {'message': 'B', 'parents': ['A']},
         {'message': 'a1'},
     ]
     repo = backend_hg.create_repo(commits=commits)
     commit_ids = backend_hg.commit_ids
 
     model = scm.ScmModel()
     model.strip(repo, commit_ids['b'], branch=None)
 
     vcs_repo = repo.scm_instance()
     rest_commit_ids = [c.raw_id for c in vcs_repo.get_changesets()]
     assert len(rest_commit_ids) == 4
     assert commit_ids['b'] not in rest_commit_ids
 
 
 def test_strip_with_single_heads(backend_hg):
     commits = [
         {'message': 'A'},
         {'message': 'a'},
         {'message': 'b'},
     ]
     repo = backend_hg.create_repo(commits=commits)
     commit_ids = backend_hg.commit_ids
 
     model = scm.ScmModel()
     model.strip(repo, commit_ids['b'], branch=None)
 
     vcs_repo = repo.scm_instance()
     rest_commit_ids = [c.raw_id for c in vcs_repo.get_changesets()]
     assert len(rest_commit_ids) == 2
     assert commit_ids['b'] not in rest_commit_ids
 
 
 def test_get_nodes_returns_unicode_flat(backend_random):
     repo = backend_random.repo
     directories, files = scm.ScmModel().get_nodes(
         repo.repo_name, repo.get_commit(commit_idx=0).raw_id,
         flat=True)
     assert_contains_only_unicode(directories)
     assert_contains_only_unicode(files)
 
 
 def test_get_nodes_returns_unicode_non_flat(backend_random):
     repo = backend_random.repo
     directories, files = scm.ScmModel().get_nodes(
         repo.repo_name, repo.get_commit(commit_idx=0).raw_id,
         flat=False)
     # johbo: Checking only the names for now, since that is the critical
     # part.
     assert_contains_only_unicode([d['name'] for d in directories])
     assert_contains_only_unicode([f['name'] for f in files])
 
 
 def test_get_nodes_max_file_bytes(backend_random):
     repo = backend_random.repo
     max_file_bytes = 10
     directories, files = scm.ScmModel().get_nodes(
         repo.repo_name, repo.get_commit(commit_idx=0).raw_id, content=True,
         extended_info=True, flat=False)
     assert any(file['content'] and len(file['content']) > max_file_bytes
                for file in files)
 
     directories, files = scm.ScmModel().get_nodes(
         repo.repo_name, repo.get_commit(commit_idx=0).raw_id, content=True,
         extended_info=True, flat=False, max_file_bytes=max_file_bytes)
     assert all(
         file['content'] is None if file['size'] > max_file_bytes else True
         for file in files)
 
 
 def assert_contains_only_unicode(structure):
     assert structure
     for value in structure:
         assert isinstance(value, unicode)
 
 
 @pytest.mark.backends("hg", "git")
 def test_get_non_unicode_reference(backend):
     model = scm.ScmModel()
     non_unicode_list = ["Adını".decode("cp1254")]
 
     def scm_instance():
         return Mock(
             branches=non_unicode_list, bookmarks=non_unicode_list,
             tags=non_unicode_list, alias=backend.alias)
 
     repo = Mock(__class__=db.Repository, scm_instance=scm_instance)
     choices, __ = model.get_repo_landing_revs(repo=repo)
     if backend.alias == 'hg':
         valid_choices = [
             'rev:tip', u'branch:Ad\xc4\xb1n\xc4\xb1',
             u'book:Ad\xc4\xb1n\xc4\xb1', u'tag:Ad\xc4\xb1n\xc4\xb1']
     else:
         valid_choices = [
             'rev:tip', u'branch:Ad\xc4\xb1n\xc4\xb1',
             u'tag:Ad\xc4\xb1n\xc4\xb1']
 
     assert choices == valid_choices
 
 
 class TestInstallSvnHooks(object):
     HOOK_FILES = ('pre-commit', 'post-commit')
 
     def test_new_hooks_are_created(self, backend_svn):
         model = scm.ScmModel()
         repo = backend_svn.create_repo()
         vcs_repo = repo.scm_instance()
         model.install_svn_hooks(vcs_repo)
 
         hooks_path = os.path.join(vcs_repo.path, 'hooks')
         assert os.path.isdir(hooks_path)
         for file_name in self.HOOK_FILES:
             file_path = os.path.join(hooks_path, file_name)
             self._check_hook_file_mode(file_path)
             self._check_hook_file_content(file_path)
 
     def test_rc_hooks_are_replaced(self, backend_svn):
         model = scm.ScmModel()
         repo = backend_svn.create_repo()
         vcs_repo = repo.scm_instance()
         hooks_path = os.path.join(vcs_repo.path, 'hooks')
         file_paths = [os.path.join(hooks_path, f) for f in self.HOOK_FILES]
 
         for file_path in file_paths:
             self._create_fake_hook(
                 file_path, content="RC_HOOK_VER = 'abcde'\n")
 
         model.install_svn_hooks(vcs_repo)
 
         for file_path in file_paths:
             self._check_hook_file_content(file_path)
 
     def test_non_rc_hooks_are_not_replaced_without_force_create(
             self, backend_svn):
         model = scm.ScmModel()
         repo = backend_svn.create_repo()
         vcs_repo = repo.scm_instance()
         hooks_path = os.path.join(vcs_repo.path, 'hooks')
         file_paths = [os.path.join(hooks_path, f) for f in self.HOOK_FILES]
         non_rc_content = "exit 0\n"
 
         for file_path in file_paths:
             self._create_fake_hook(file_path, content=non_rc_content)
 
         model.install_svn_hooks(vcs_repo)
 
         for file_path in file_paths:
             with open(file_path, 'rt') as hook_file:
                 content = hook_file.read()
             assert content == non_rc_content
 
     def test_non_rc_hooks_are_replaced_with_force_create(self, backend_svn):
         model = scm.ScmModel()
         repo = backend_svn.create_repo()
         vcs_repo = repo.scm_instance()
         hooks_path = os.path.join(vcs_repo.path, 'hooks')
         file_paths = [os.path.join(hooks_path, f) for f in self.HOOK_FILES]
         non_rc_content = "exit 0\n"
 
         for file_path in file_paths:
             self._create_fake_hook(file_path, content=non_rc_content)
 
         model.install_svn_hooks(vcs_repo, force_create=True)
 
         for file_path in file_paths:
             self._check_hook_file_content(file_path)
 
     def _check_hook_file_mode(self, file_path):
         assert os.path.exists(file_path)
         stat_info = os.stat(file_path)
 
         file_mode = stat.S_IMODE(stat_info.st_mode)
         expected_mode = int('755', 8)
         assert expected_mode == file_mode
 
     def _check_hook_file_content(self, file_path):
         with open(file_path, 'rt') as hook_file:
             content = hook_file.read()
 
         expected_env = '#!{}'.format(sys.executable)
         expected_rc_version = "\nRC_HOOK_VER = '{}'\n".format(
             rhodecode.__version__)
         assert content.strip().startswith(expected_env)
         assert expected_rc_version in content
 
     def _create_fake_hook(self, file_path, content):
         with open(file_path, 'w') as hook_file:
             hook_file.write(content)
 
 
 class TestCheckRhodecodeHook(object):
 
     @patch('os.path.exists', Mock(return_value=False))
     def test_returns_true_when_no_hook_found(self):
         result = scm._check_rhodecode_hook('/tmp/fake_hook_file.py')
         assert result
 
     @pytest.mark.parametrize("file_content, expected_result", [
         ("RC_HOOK_VER = '3.3.3'\n", True),
         ("RC_HOOK = '3.3.3'\n", False),
-    ])
+    ], ids=no_newline_id_generator)
     @patch('os.path.exists', Mock(return_value=True))
     def test_signatures(self, file_content, expected_result):
         hook_content_patcher = patch.object(
             scm, '_read_hook', return_value=file_content)
         with hook_content_patcher:
             result = scm._check_rhodecode_hook('/tmp/fake_hook_file.py')
 
         assert result is expected_result
 
 
 class TestInstallHooks(object):
     def test_hooks_are_installed_for_git_repo(self, backend_git):
         repo = backend_git.create_repo()
         model = scm.ScmModel()
         scm_repo = repo.scm_instance()
         with patch.object(model, 'install_git_hook') as hooks_mock:
             model.install_hooks(scm_repo, repo_type='git')
         hooks_mock.assert_called_once_with(scm_repo)
 
     def test_hooks_are_installed_for_svn_repo(self, backend_svn):
         repo = backend_svn.create_repo()
         scm_repo = repo.scm_instance()
         model = scm.ScmModel()
         with patch.object(scm.ScmModel, 'install_svn_hooks') as hooks_mock:
             model.install_hooks(scm_repo, repo_type='svn')
         hooks_mock.assert_called_once_with(scm_repo)
 
     @pytest.mark.parametrize('hook_method', [
         'install_svn_hooks',
         'install_git_hook'])
     def test_mercurial_doesnt_trigger_hooks(self, backend_hg, hook_method):
         repo = backend_hg.create_repo()
         scm_repo = repo.scm_instance()
         model = scm.ScmModel()
         with patch.object(scm.ScmModel, hook_method) as hooks_mock:
             model.install_hooks(scm_repo, repo_type='hg')
         assert hooks_mock.call_count == 0

rhodecode/tests/other/test_libs.py

@@ -1,527 +1,529 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/
 
 
 """
 Package for testing various lib/helper functions in rhodecode
 """
 
 import datetime
 import string
 import mock
 import pytest
+
+from rhodecode.tests import no_newline_id_generator
 from rhodecode.tests.utils import run_test_concurrently
 from rhodecode.lib.helpers import InitialsGravatar
 
 from rhodecode.lib.utils2 import AttributeDict
 from rhodecode.model.db import Repository
 
 
 def _urls_for_proto(proto):
     return [
         ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
          '%s://127.0.0.1' % proto),
         ('%s://marcink@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
          '%s://127.0.0.1' % proto),
         ('%s://marcink:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
          '%s://127.0.0.1' % proto),
         ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],
          '%s://127.0.0.1:8080' % proto),
         ('%s://domain.org' % proto, ['%s://' % proto, 'domain.org'],
          '%s://domain.org' % proto),
         ('%s://user:[e-mail]:8080' % proto,
          ['%s://' % proto, 'domain.org', '8080'],
          '%s://domain.org:8080' % proto),
     ]
 
 TEST_URLS = _urls_for_proto('http') + _urls_for_proto('https')
 
 
 @pytest.mark.parametrize("test_url, expected, expected_creds", TEST_URLS)
 def test_uri_filter(test_url, expected, expected_creds):
     from rhodecode.lib.utils2 import uri_filter
     assert uri_filter(test_url) == expected
 
 
 @pytest.mark.parametrize("test_url, expected, expected_creds", TEST_URLS)
 def test_credentials_filter(test_url, expected, expected_creds):
     from rhodecode.lib.utils2 import credentials_filter
     assert credentials_filter(test_url) == expected_creds
 
 
 @pytest.mark.parametrize("str_bool, expected", [
     ('t', True),
     ('true', True),
     ('y', True),
     ('yes', True),
     ('on', True),
     ('1', True),
     ('Y', True),
     ('yeS', True),
     ('Y', True),
     ('TRUE', True),
     ('T', True),
     ('False', False),
     ('F', False),
     ('FALSE', False),
     ('0', False),
     ('-1', False),
     ('', False)
 ])
 def test_str2bool(str_bool, expected):
     from rhodecode.lib.utils2 import str2bool
     assert str2bool(str_bool) == expected
 
 
 @pytest.mark.parametrize("text, expected", reduce(lambda a1,a2:a1+a2, [
 [
 (pref+"", []),
 (pref+"Hi there @marcink", ['marcink']),
 (pref+"Hi there @marcink and @bob", ['bob', 'marcink']),
 (pref+"Hi there @marcink\n", ['marcink']),
 (pref+"Hi there @marcink and @bob\n", ['bob', 'marcink']),
 (pref+"Hi there [e-mail]", []),
 (pref+"Hi there @john.malcovic and @bob\n", ['bob', 'john.malcovic']),
 (pref+"This needs to be reviewed: (@marcink,@john)", ["john", "marcink"]),
 (pref+"This needs to be reviewed: (@marcink, @john)", ["john", "marcink"]),
 (pref+"This needs to be reviewed: [@marcink,@john]", ["john", "marcink"]),
 (pref+"This needs to be reviewed: (@marcink @john)", ["john", "marcink"]),
 (pref+"@john @mary, please review", ["john", "mary"]),
 (pref+"@john,@mary, please review", ["john", "mary"]),
 (pref+"Hej @123, @22john,@mary, please review", ['123', '22john', 'mary']),
 (pref+"@first hi there @marcink here's my email [e-mail] "
  "@lukaszb check @one_more22 it pls @ ttwelve @D[] @one@two@three ", ['first', 'lukaszb', 'marcink', 'one', 'one_more22']),
 (pref+"@MARCIN    @maRCiN @2one_more22 @john please see this http://org.pl", ['2one_more22', 'john', 'MARCIN', 'maRCiN']),
 (pref+"@marian.user just do it @marco-polo and next extract @marco_polo", ['marco-polo', 'marco_polo', 'marian.user']),
 (pref+"user.dot  hej ! not-needed [e-mail]", []),
 (pref+"\n@marcin", ['marcin']),
 ]
-for pref in ['', '\n', 'hi !', '\t', '\n\n']]))
+for pref in ['', '\n', 'hi !', '\t', '\n\n']]), ids=no_newline_id_generator)
 def test_mention_extractor(text, expected):
     from rhodecode.lib.utils2 import extract_mentioned_users
     got = extract_mentioned_users(text)
     assert sorted(got, key=lambda x: x.lower()) == got
     assert set(expected) == set(got)
 
 @pytest.mark.parametrize("age_args, expected, kw", [
     ({}, u'just now', {}),
     ({'seconds': -1}, u'1 second ago', {}),
     ({'seconds': -60 * 2}, u'2 minutes ago', {}),
     ({'hours': -1}, u'1 hour ago', {}),
     ({'hours': -24}, u'1 day ago', {}),
     ({'hours': -24 * 5}, u'5 days ago', {}),
     ({'months': -1}, u'1 month ago', {}),
     ({'months': -1, 'days': -2}, u'1 month and 2 days ago', {}),
     ({'years': -1, 'months': -1}, u'1 year and 1 month ago', {}),
     ({}, u'just now', {'short_format': True}),
     ({'seconds': -1}, u'1sec ago', {'short_format': True}),
     ({'seconds': -60 * 2}, u'2min ago', {'short_format': True}),
     ({'hours': -1}, u'1h ago', {'short_format': True}),
     ({'hours': -24}, u'1d ago', {'short_format': True}),
     ({'hours': -24 * 5}, u'5d ago', {'short_format': True}),
     ({'months': -1}, u'1m ago', {'short_format': True}),
     ({'months': -1, 'days': -2}, u'1m, 2d ago', {'short_format': True}),
     ({'years': -1, 'months': -1}, u'1y, 1m ago', {'short_format': True}),
 ])
 def test_age(age_args, expected, kw, pylonsapp):
     from rhodecode.lib.utils2 import age
     from dateutil import relativedelta
     n = datetime.datetime(year=2012, month=5, day=17)
     delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
 
     def translate(elem):
         return elem.interpolate()
 
     assert translate(age(n + delt(**age_args), now=n, **kw)) == expected
 
 
 @pytest.mark.parametrize("age_args, expected, kw", [
     ({}, u'just now', {}),
     ({'seconds': 1}, u'in 1 second', {}),
     ({'seconds': 60 * 2}, u'in 2 minutes', {}),
     ({'hours': 1}, u'in 1 hour', {}),
     ({'hours': 24}, u'in 1 day', {}),
     ({'hours': 24 * 5}, u'in 5 days', {}),
     ({'months': 1}, u'in 1 month', {}),
     ({'months': 1, 'days': 1}, u'in 1 month and 1 day', {}),
     ({'years': 1, 'months': 1}, u'in 1 year and 1 month', {}),
     ({}, u'just now', {'short_format': True}),
     ({'seconds': 1}, u'in 1sec', {'short_format': True}),
     ({'seconds': 60 * 2}, u'in 2min', {'short_format': True}),
     ({'hours': 1}, u'in 1h', {'short_format': True}),
     ({'hours': 24}, u'in 1d', {'short_format': True}),
     ({'hours': 24 * 5}, u'in 5d', {'short_format': True}),
     ({'months': 1}, u'in 1m', {'short_format': True}),
     ({'months': 1, 'days': 1}, u'in 1m, 1d', {'short_format': True}),
     ({'years': 1, 'months': 1}, u'in 1y, 1m', {'short_format': True}),
 ])
 def test_age_in_future(age_args, expected, kw, pylonsapp):
     from rhodecode.lib.utils2 import age
     from dateutil import relativedelta
     n = datetime.datetime(year=2012, month=5, day=17)
     delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
 
     def translate(elem):
         return elem.interpolate()
 
     assert translate(age(n + delt(**age_args), now=n, **kw)) == expected
 
 
 def test_tag_exctrator():
     sample = (
         "hello pta[tag] gog [[]] [[] sda ero[or]d [me =>>< sa]"
         "[requires] [stale] [see<>=>] [see => http://url.com]"
         "[requires => url] [lang => python] [just a tag] <html_tag first='abc' attr=\"my.url?attr=&another=\"></html_tag>"
         "[,d] [ => ULR ] [obsolete] [desc]]"
     )
     from rhodecode.lib.helpers import desc_stylize, escaped_stylize
     res = desc_stylize(sample)
     assert '<div class="metatag" tag="tag">tag</div>' in res
     assert '<div class="metatag" tag="obsolete">obsolete</div>' in res
     assert '<div class="metatag" tag="stale">stale</div>' in res
     assert '<div class="metatag" tag="lang">python</div>' in res
     assert '<div class="metatag" tag="requires">requires =&gt; <a href="/url">url</a></div>' in res
     assert '<div class="metatag" tag="tag">tag</div>' in res
     assert '<html_tag first=\'abc\' attr=\"my.url?attr=&another=\"></html_tag>' in res
 
     res_encoded = escaped_stylize(sample)
     assert '<div class="metatag" tag="tag">tag</div>' in res_encoded
     assert '<div class="metatag" tag="obsolete">obsolete</div>' in res_encoded
     assert '<div class="metatag" tag="stale">stale</div>' in res_encoded
     assert '<div class="metatag" tag="lang">python</div>' in res_encoded
     assert '<div class="metatag" tag="requires">requires =&gt; <a href="/url">url</a></div>' in res_encoded
     assert '<div class="metatag" tag="tag">tag</div>' in res_encoded
     assert '&lt;html_tag first=&#39;abc&#39; attr=&#34;my.url?attr=&amp;another=&#34;&gt;&lt;/html_tag&gt;' in res_encoded
 
 
 @pytest.mark.parametrize("tmpl_url, email, expected", [
     ('http://test.com/{email}', '[e-mail]', 'http://test.com/[e-mail]'),
 
     ('http://test.com/{md5email}', '[e-mail]', 'http://test.com/3cb7232fcc48743000cb86d0d5022bd9'),
     ('http://test.com/{md5email}', 'testąć@foo.com', 'http://test.com/978debb907a3c55cd741872ab293ef30'),
 
     ('http://testX.com/{md5email}?s={size}', '[e-mail]', 'http://testX.com/3cb7232fcc48743000cb86d0d5022bd9?s=24'),
     ('http://testX.com/{md5email}?s={size}', 'testąć@foo.com', 'http://testX.com/978debb907a3c55cd741872ab293ef30?s=24'),
 
     ('{scheme}://{netloc}/{md5email}/{size}', '[e-mail]', 'https://server.com/3cb7232fcc48743000cb86d0d5022bd9/24'),
     ('{scheme}://{netloc}/{md5email}/{size}', 'testąć@foo.com', 'https://server.com/978debb907a3c55cd741872ab293ef30/24'),
 
     ('http://test.com/{email}', 'testąć@foo.com', 'http://test.com/testąć@foo.com'),
     ('http://test.com/{email}?size={size}', '[e-mail]', 'http://test.com/[e-mail]?size=24'),
     ('http://test.com/{email}?size={size}', 'testąć@foo.com', 'http://test.com/testąć@foo.com?size=24'),
 ])
 def test_gravatar_url_builder(tmpl_url, email, expected, request_stub):
     from rhodecode.lib.helpers import gravatar_url
 
     # mock pyramid.threadlocals
     def fake_get_current_request():
         request_stub.scheme = 'https'
         request_stub.host = 'server.com'
         return request_stub
 
     # mock pylons.tmpl_context
     def fake_tmpl_context(_url):
         _c = AttributeDict()
         _c.visual = AttributeDict()
         _c.visual.use_gravatar = True
         _c.visual.gravatar_url = _url
 
         return _c
 
     with mock.patch('rhodecode.lib.helpers.get_current_request',
                     fake_get_current_request):
         fake = fake_tmpl_context(_url=tmpl_url)
         with mock.patch('pylons.tmpl_context', fake):
             grav = gravatar_url(email_address=email, size=24)
             assert grav == expected
 
 
 @pytest.mark.parametrize(
     "email, first_name, last_name, expected_initials, expected_color", [
 
         ('[e-mail]', '', '', 'TR', '#8a994d'),
         ('[e-mail]', '', '', 'MK', '#6559b3'),
         # special cases of email
         ('[e-mail]', '', '', 'JD', '#526600'),
         ('[e-mail]', '', '', 'GR', '#990052'),
         ('[e-mail]', 'Guido', 'Van Rossum', 'GR', '#990052'),
 
         ('[e-mail]', '', '', 'RR', '#46598c'),
         ('[e-mail]', 'Nguyễn Thái', 'Tgọc Duy', 'ND', '#665200'),
 
         ('[e-mail]', '', '', 'JF', '#73006b'),
         ('[e-mail]', 'Marcin', 'Kuzminski', 'MK', '#104036'),
         # partials
         ('[e-mail]', 'Marcin', '', 'MR', '#104036'),  # fn+email
         ('[e-mail]', '', 'Kuzminski', 'AK', '#104036'),  # em+ln
         # non-ascii
         ('[e-mail]', 'Marcin', 'Śuzminski', 'MS', '#104036'),
         ('marcin.ś[e-mail]', '', '', 'MS', '#73000f'),
 
         # special cases, LDAP can provide those...
         ('admin@', 'Marcin', 'Śuzminski', 'MS', '#aa00ff'),
         ('marcin.śuzminski', '', '', 'MS', '#402020'),
         ('null', '', '', 'NL', '#8c4646'),
 ])
 def test_initials_gravatar_pick_of_initials_and_color_algo(
         email, first_name, last_name, expected_initials, expected_color):
     instance = InitialsGravatar(email, first_name, last_name)
     assert instance.get_initials() == expected_initials
     assert instance.str2color(email) == expected_color
 
 
 def test_initials_gravatar_mapping_algo():
     pos = set()
     instance = InitialsGravatar('', '', '')
     iterations = 0
 
     variations = []
     for letter1 in string.ascii_letters:
         for letter2 in string.ascii_letters[::-1][:10]:
             for letter3 in string.ascii_letters[:10]:
                 variations.append(
                     '[e-mail]' % (letter1+letter2+letter3))
 
     max_variations = 4096
     for email in variations[:max_variations]:
         iterations += 1
         pos.add(
             instance.pick_color_bank_index(email,
                                            instance.get_color_bank()))
 
     # we assume that we have match all 256 possible positions,
     # in reasonable amount of different email addresses
     assert len(pos) == 256
     assert iterations == max_variations
 
 
 @pytest.mark.parametrize("tmpl, repo_name, overrides, prefix, expected", [
     (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '', 'http://vps1:8000/group/repo1'),
     (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'marcink'}, '', 'http://marcink@vps1:8000/group/repo1'),
     (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '/rc', 'http://vps1:8000/rc/group/repo1'),
     (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/rc', 'http://user@vps1:8000/rc/group/repo1'),
     (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'marcink'}, '/rc', 'http://marcink@vps1:8000/rc/group/repo1'),
     (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/rc/', 'http://user@vps1:8000/rc/group/repo1'),
     (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'marcink'}, '/rc/', 'http://marcink@vps1:8000/rc/group/repo1'),
     ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {}, '', 'http://vps1:8000/_23'),
     ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'marcink'}, '', 'http://marcink@vps1:8000/_23'),
     ('http://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'marcink'}, '', 'http://marcink@vps1:8000/_23'),
     ('http://{netloc}/_{repoid}', 'group/repo1', {'user': 'marcink'}, '', 'http://vps1:8000/_23'),
     ('https://{user}@proxy1.server.com/{repo}', 'group/repo1', {'user': 'marcink'}, '', 'https://[e-mail]/group/repo1'),
     ('https://{user}@proxy1.server.com/{repo}', 'group/repo1', {}, '', 'https://proxy1.server.com/group/repo1'),
     ('https://proxy1.server.com/{user}/{repo}', 'group/repo1', {'user': 'marcink'}, '', 'https://proxy1.server.com/marcink/group/repo1'),
 ])
 def test_clone_url_generator(tmpl, repo_name, overrides, prefix, expected):
     from rhodecode.lib.utils2 import get_clone_url
     clone_url = get_clone_url(uri_tmpl=tmpl, qualifed_home_url='http://vps1:8000'+prefix,
                               repo_name=repo_name, repo_id=23, **overrides)
     assert clone_url == expected
 
 
 def _quick_url(text, tmpl="""<a class="revision-link" href="%s">%s</a>""", url_=None):
     """
     Changes `some text url[foo]` => `some text <a href="/">foo</a>
 
     :param text:
     """
     import re
     # quickly change expected url[] into a link
     URL_PAT = re.compile(r'(?:url\[)(.+?)(?:\])')
 
     def url_func(match_obj):
         _url = match_obj.groups()[0]
         return tmpl % (url_ or '/some-url', _url)
     return URL_PAT.sub(url_func, text)
 
 
 @pytest.mark.parametrize("sample, expected", [
   ("",
    ""),
   ("git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68",
    "git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68"),
   ("from rev 000000000000",
    "from rev url[000000000000]"),
   ("from rev 000000000000123123 also rev 000000000000",
    "from rev url[000000000000123123] also rev url[000000000000]"),
   ("this should-000 00",
    "this should-000 00"),
   ("longtextffffffffff rev 123123123123",
    "longtextffffffffff rev url[123123123123]"),
   ("rev ffffffffffffffffffffffffffffffffffffffffffffffffff",
    "rev ffffffffffffffffffffffffffffffffffffffffffffffffff"),
   ("ffffffffffff some text traalaa",
    "url[ffffffffffff] some text traalaa"),
    ("""Multi line
    123123123123
    some text 123123123123
    sometimes !
    """,
    """Multi line
    url[123123123123]
    some text url[123123123123]
    sometimes !
    """)
-])
+], ids=no_newline_id_generator)
 def test_urlify_commits(sample, expected):
     def fake_url(self, *args, **kwargs):
         return '/some-url'
 
     expected = _quick_url(expected)
 
     with mock.patch('pylons.url', fake_url):
         from rhodecode.lib.helpers import urlify_commits
         assert urlify_commits(sample, 'repo_name') == expected
 
 
 @pytest.mark.parametrize("sample, expected, url_", [
     ("",
      "",
      ""),
     ("https://svn.apache.org/repos",
      "url[https://svn.apache.org/repos]",
      "https://svn.apache.org/repos"),
     ("http://svn.apache.org/repos",
      "url[http://svn.apache.org/repos]",
      "http://svn.apache.org/repos"),
     ("from rev a also rev http://google.com",
      "from rev a also rev url[http://google.com]",
      "http://google.com"),
     ("""Multi line
      https://foo.bar.com
      some text lalala""",
      """Multi line
      url[https://foo.bar.com]
      some text lalala""",
      "https://foo.bar.com")
-])
+], ids=no_newline_id_generator)
 def test_urlify_test(sample, expected, url_):
     from rhodecode.lib.helpers import urlify_text
     expected = _quick_url(expected, tmpl="""<a href="%s">%s</a>""", url_=url_)
     assert urlify_text(sample) == expected
 
 
 @pytest.mark.parametrize("test, expected", [
   ("", None),
   ("/_2", '2'),
   ("_2", '2'),
   ("/_2/", '2'),
   ("_2/", '2'),
 
   ("/_21", '21'),
   ("_21", '21'),
   ("/_21/", '21'),
   ("_21/", '21'),
 
   ("/_21/foobar", '21'),
   ("_21/121", '21'),
   ("/_21/_12", '21'),
   ("_21/rc/foo", '21'),
 
 ])
 def test_get_repo_by_id(test, expected):
     from rhodecode.model.repo import RepoModel
     _test = RepoModel()._extract_id_from_repo_name(test)
     assert _test == expected
 
 
 @pytest.mark.parametrize("test_repo_name, repo_type", [
   ("test_repo_1", None),
   ("repo_group/foobar", None),
   ("test_non_asci_ąćę", None),
   (u"test_non_asci_unicode_ąćę", None),
 ])
 def test_invalidation_context(pylonsapp, test_repo_name, repo_type):
     from beaker.cache import cache_region
     from rhodecode.lib import caches
     from rhodecode.model.db import CacheKey
 
     @cache_region('long_term')
     def _dummy_func(cache_key):
         return 'result'
 
     invalidator_context = CacheKey.repo_context_cache(
         _dummy_func, test_repo_name, 'repo')
 
     with invalidator_context as context:
         invalidated = context.invalidate()
         result = context.compute()
 
     assert invalidated == True
     assert 'result' == result
     assert isinstance(context, caches.FreshRegionCache)
 
     assert 'InvalidationContext' in repr(invalidator_context)
 
     with invalidator_context as context:
         context.invalidate()
         result = context.compute()
 
     assert 'result' == result
     assert isinstance(context, caches.ActiveRegionCache)
 
 
 def test_invalidation_context_exception_in_compute(pylonsapp):
     from rhodecode.model.db import CacheKey
     from beaker.cache import cache_region
 
     @cache_region('long_term')
     def _dummy_func(cache_key):
         # this causes error since it doesn't get any params
         raise Exception('ups')
 
     invalidator_context = CacheKey.repo_context_cache(
         _dummy_func, 'test_repo_2', 'repo')
 
     with pytest.raises(Exception):
         with invalidator_context as context:
             context.invalidate()
             context.compute()
 
 
 @pytest.mark.parametrize('execution_number', range(5))
 def test_cache_invalidation_race_condition(execution_number, pylonsapp):
     import time
     from beaker.cache import cache_region
     from rhodecode.model.db import CacheKey
 
     if CacheKey.metadata.bind.url.get_backend_name() == "mysql":
         reason = (
             'Fails on MariaDB due to some locking issues. Investigation'
             ' needed')
         pytest.xfail(reason=reason)
 
     @run_test_concurrently(25)
     def test_create_and_delete_cache_keys():
         time.sleep(0.2)
 
         @cache_region('long_term')
         def _dummy_func(cache_key):
             return 'result'
 
         invalidator_context = CacheKey.repo_context_cache(
             _dummy_func, 'test_repo_1', 'repo')
 
         with invalidator_context as context:
             context.invalidate()
             context.compute()
 
         CacheKey.set_invalidate('test_repo_1', delete=True)
 
     test_create_and_delete_cache_keys()