##// END OF EJS Templates
perms-summary: fix long names overflow....
perms-summary: fix long names overflow. - fixes #5336

File last commit:

r1985:f5a73a5e default
r2008:5fbc0fc6 default
Show More
test_admin_repos.py
1131 lines | 45.8 KiB | text/x-python | PythonLexer
# -*- 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
from rhodecode.lib import helpers as h
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, logout_user_session)
from rhodecode.tests.fixture import Fixture, error_function
from rhodecode.tests.utils import AssertResponse, repo_on_filesystem
fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib
base_url = {
'repo_summary': '/{repo_name}',
'repo_creating_check': '/{repo_name}/repo_creating_check',
}[name].format(**kwargs)
if params:
base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
return base_url
@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')
def test_show(self, autologin_user, backend):
self.app.get(url('repo', repo_name=backend.repo_name))
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_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(
route_path('repo_creating_check', repo_name=safe_str(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(
h.route_path('repo_summary', repo_name=safe_str(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,
'extensions_evolve': 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