engineering/unrhodecode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

repo-group-schemas: refactor repository group schemas.

Browse source 
- fixes #4133
- refactor to make it behave like repo schemas
- chain validation fixes
- nicer error messages
- consistent permission checks
This commit is contained in:
Marcin Kuzminski 2016-11-24 23:11:23 +01:00
parent c081457f2c
commit fb621ee72c
5 changed files with 636 additions and 173 deletions
Download patch file Download diff file Expand all files Collapse all files

200
rhodecode/api/tests/test_create_repo_group.py
View file

@ -54,55 +54,10 @@ class TestCreateRepoGroup(object):
'repo_group': repo_group.get_api_data()
}
expected = ret
assert_ok(id_, expected, given=response.body)
fixture.destroy_repo_group(repo_group_name)
def test_api_create_repo_group_regular_user(self):
repo_group_name = 'api-repo-group'
usr = UserModel().get_by_username(self.TEST_USER_LOGIN)
usr.inherit_default_permissions = False
Session().add(usr)
UserModel().grant_perm(
self.TEST_USER_LOGIN, 'hg.repogroup.create.true')
Session().commit()
repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
assert repo_group is None
id_, params = build_data(
self.apikey_regular, 'create_repo_group',
group_name=repo_group_name,
owner=TEST_USER_ADMIN_LOGIN,)
response = api_call(self.app, params)
repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
assert repo_group is not None
ret = {
'msg': 'Created new repo group `%s`' % (repo_group_name,),
'repo_group': repo_group.get_api_data()
}
expected = ret
assert_ok(id_, expected, given=response.body)
fixture.destroy_repo_group(repo_group_name)
UserModel().revoke_perm(
self.TEST_USER_LOGIN, 'hg.repogroup.create.true')
usr = UserModel().get_by_username(self.TEST_USER_LOGIN)
usr.inherit_default_permissions = True
Session().add(usr)
Session().commit()
def test_api_create_repo_group_regular_user_no_permission(self):
repo_group_name = 'api-repo-group'
id_, params = build_data(
self.apikey_regular, 'create_repo_group',
group_name=repo_group_name,
owner=TEST_USER_ADMIN_LOGIN,)
response = api_call(self.app, params)
expected = "Access was denied to this resource."
assert_error(id_, expected, given=response.body)
try:
assert_ok(id_, expected, given=response.body)
finally:
fixture.destroy_repo_group(repo_group_name)
def test_api_create_repo_group_in_another_group(self):
repo_group_name = 'api-repo-group'
@ -127,9 +82,11 @@ class TestCreateRepoGroup(object):
'repo_group': repo_group.get_api_data()
}
expected = ret
assert_ok(id_, expected, given=response.body)
fixture.destroy_repo_group(full_repo_group_name)
fixture.destroy_repo_group(repo_group_name)
try:
assert_ok(id_, expected, given=response.body)
finally:
fixture.destroy_repo_group(full_repo_group_name)
fixture.destroy_repo_group(repo_group_name)
def test_api_create_repo_group_in_another_group_not_existing(self):
repo_group_name = 'api-repo-group-no'
@ -144,7 +101,10 @@ class TestCreateRepoGroup(object):
owner=TEST_USER_ADMIN_LOGIN,
copy_permissions=True)
response = api_call(self.app, params)
expected = 'repository group `%s` does not exist' % (repo_group_name,)
expected = {
'repo_group':
'Parent repository group `{}` does not exist'.format(
repo_group_name)}
assert_error(id_, expected, given=response.body)
def test_api_create_repo_group_that_exists(self):
@ -159,9 +119,139 @@ class TestCreateRepoGroup(object):
group_name=repo_group_name,
owner=TEST_USER_ADMIN_LOGIN,)
response = api_call(self.app, params)
expected = 'repo group `%s` already exist' % (repo_group_name,)
expected = {
'unique_repo_group_name':
'Repository group with name `{}` already exists'.format(
repo_group_name)}
try:
assert_error(id_, expected, given=response.body)
finally:
fixture.destroy_repo_group(repo_group_name)
def test_api_create_repo_group_regular_user_wit_root_location_perms(
self, user_util):
regular_user = user_util.create_user()
regular_user_api_key = regular_user.api_key
repo_group_name = 'api-repo-group-by-regular-user'
usr = UserModel().get_by_username(regular_user.username)
usr.inherit_default_permissions = False
Session().add(usr)
UserModel().grant_perm(
regular_user.username, 'hg.repogroup.create.true')
Session().commit()
repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
assert repo_group is None
id_, params = build_data(
regular_user_api_key, 'create_repo_group',
group_name=repo_group_name)
response = api_call(self.app, params)
repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
assert repo_group is not None
expected = {
'msg': 'Created new repo group `%s`' % (repo_group_name,),
'repo_group': repo_group.get_api_data()
}
try:
assert_ok(id_, expected, given=response.body)
finally:
fixture.destroy_repo_group(repo_group_name)
def test_api_create_repo_group_regular_user_with_admin_perms_to_parent(
self, user_util):
repo_group_name = 'api-repo-group-parent'
repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
assert repo_group is None
# create the parent
fixture.create_repo_group(repo_group_name)
# user perms
regular_user = user_util.create_user()
regular_user_api_key = regular_user.api_key
usr = UserModel().get_by_username(regular_user.username)
usr.inherit_default_permissions = False
Session().add(usr)
RepoGroupModel().grant_user_permission(
repo_group_name, regular_user.username, 'group.admin')
Session().commit()
full_repo_group_name = repo_group_name + '/' + repo_group_name
id_, params = build_data(
regular_user_api_key, 'create_repo_group',
group_name=full_repo_group_name)
response = api_call(self.app, params)
repo_group = RepoGroupModel.cls.get_by_group_name(full_repo_group_name)
assert repo_group is not None
expected = {
'msg': 'Created new repo group `{}`'.format(full_repo_group_name),
'repo_group': repo_group.get_api_data()
}
try:
assert_ok(id_, expected, given=response.body)
finally:
fixture.destroy_repo_group(full_repo_group_name)
fixture.destroy_repo_group(repo_group_name)
def test_api_create_repo_group_regular_user_no_permission_to_create_to_root_level(self):
repo_group_name = 'api-repo-group'
id_, params = build_data(
self.apikey_regular, 'create_repo_group',
group_name=repo_group_name)
response = api_call(self.app, params)
expected = {
'repo_group':
u'You do not have the permission to store '
u'repository groups in the root location.'}
assert_error(id_, expected, given=response.body)
def test_api_create_repo_group_regular_user_no_parent_group_perms(self):
repo_group_name = 'api-repo-group-regular-user'
repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
assert repo_group is None
# create the parent
fixture.create_repo_group(repo_group_name)
full_repo_group_name = repo_group_name+'/'+repo_group_name
id_, params = build_data(
self.apikey_regular, 'create_repo_group',
group_name=full_repo_group_name)
response = api_call(self.app, params)
expected = {
'repo_group':
'Parent repository group `{}` does not exist'.format(
repo_group_name)}
try:
assert_error(id_, expected, given=response.body)
finally:
fixture.destroy_repo_group(repo_group_name)
def test_api_create_repo_group_regular_user_no_permission_to_specify_owner(
self):
repo_group_name = 'api-repo-group'
id_, params = build_data(
self.apikey_regular, 'create_repo_group',
group_name=repo_group_name,
owner=TEST_USER_ADMIN_LOGIN,)
response = api_call(self.app, params)
expected = "Only RhodeCode super-admin can specify `owner` param"
assert_error(id_, expected, given=response.body)
fixture.destroy_repo_group(repo_group_name)
@mock.patch.object(RepoGroupModel, 'create', crash)
def test_api_create_repo_group_exception_occurred(self):

69
rhodecode/api/tests/test_update_repo_group.py
View file

@ -30,22 +30,30 @@ from rhodecode.api.tests.utils import (
@pytest.mark.usefixtures("testuser_api", "app")
class TestApiUpdateRepoGroup(object):
def test_update_group_name(self, user_util):
new_group_name = 'new-group'
initial_name = self._update(user_util, group_name=new_group_name)
assert RepoGroupModel()._get_repo_group(initial_name) is None
assert RepoGroupModel()._get_repo_group(new_group_name) is not None
new_group = RepoGroupModel()._get_repo_group(new_group_name)
assert new_group is not None
assert new_group.full_path == new_group_name
def test_update_group_name_change_parent(self, user_util):
def test_update_parent(self, user_util):
parent_group = user_util.create_repo_group()
initial_name = self._update(user_util, parent=parent_group.name)
parent_group_name = parent_group.name
expected_group_name = '{}/{}'.format(parent_group_name, 'new-group')
initial_name = self._update(user_util, group_name=expected_group_name)
expected_group_name = '{}/{}'.format(parent_group.name, initial_name)
repo_group = RepoGroupModel()._get_repo_group(expected_group_name)
assert repo_group is not None
assert repo_group.group_name == expected_group_name
assert repo_group.name == initial_name
assert repo_group.full_path == expected_group_name
assert RepoGroupModel()._get_repo_group(initial_name) is None
new_path = os.path.join(
RepoGroupModel().repos_path, *repo_group.full_path_splitted)
assert os.path.exists(new_path)
@ -67,15 +75,47 @@ class TestApiUpdateRepoGroup(object):
repo_group = RepoGroupModel()._get_repo_group(initial_name)
assert repo_group.user.username == owner
def test_api_update_repo_group_by_regular_user_no_permission(
self, backend):
repo = backend.create_repo()
repo_name = repo.repo_name
def test_update_group_name_conflict_with_existing(self, user_util):
group_1 = user_util.create_repo_group()
group_2 = user_util.create_repo_group()
repo_group_name_1 = group_1.group_name
repo_group_name_2 = group_2.group_name
id_, params = build_data(
self.apikey_regular, 'update_repo_group', repogroupid=repo_name)
self.apikey, 'update_repo_group', repogroupid=repo_group_name_1,
group_name=repo_group_name_2)
response = api_call(self.app, params)
expected = 'repository group `%s` does not exist' % (repo_name,)
expected = {
'unique_repo_group_name':
'Repository group with name `{}` already exists'.format(
repo_group_name_2)}
assert_error(id_, expected, given=response.body)
def test_api_update_repo_group_by_regular_user_no_permission(self, user_util):
temp_user = user_util.create_user()
temp_user_api_key = temp_user.api_key
parent_group = user_util.create_repo_group()
repo_group_name = parent_group.group_name
id_, params = build_data(
temp_user_api_key, 'update_repo_group', repogroupid=repo_group_name)
response = api_call(self.app, params)
expected = 'repository group `%s` does not exist' % (repo_group_name,)
assert_error(id_, expected, given=response.body)
def test_api_update_repo_group_regular_user_no_root_write_permissions(
self, user_util):
temp_user = user_util.create_user()
temp_user_api_key = temp_user.api_key
parent_group = user_util.create_repo_group(owner=temp_user.username)
repo_group_name = parent_group.group_name
id_, params = build_data(
temp_user_api_key, 'update_repo_group', repogroupid=repo_group_name,
group_name='at-root-level')
response = api_call(self.app, params)
expected = {
'repo_group': 'You do not have the permission to store '
'repository groups in the root location.'}
assert_error(id_, expected, given=response.body)
def _update(self, user_util, **kwargs):
@ -89,7 +129,10 @@ class TestApiUpdateRepoGroup(object):
self.apikey, 'update_repo_group', repogroupid=initial_name,
**kwargs)
response = api_call(self.app, params)
ret = {
repo_group = RepoGroupModel.cls.get(repo_group.group_id)
expected = {
'msg': 'updated repository group ID:{} {}'.format(
repo_group.group_id, repo_group.group_name),
'repo_group': {
@ -103,5 +146,5 @@ class TestApiUpdateRepoGroup(object):
if repo_group.parent_group else None)
}
}
assert_ok(id_, ret, given=response.body)
assert_ok(id_, expected, given=response.body)
return initial_name

209
rhodecode/api/views/repo_group_api.py
View file

@ -21,19 +21,18 @@
import logging
import colander
from rhodecode.api import jsonrpc_method, JSONRPCError, JSONRPCForbidden
from rhodecode.api import JSONRPCValidationError
from rhodecode.api import jsonrpc_method, JSONRPCError
from rhodecode.api.utils import (
has_superadmin_permission, Optional, OAttr, get_user_or_error,
store_update, get_repo_group_or_error,
get_perm_or_error, get_user_group_or_error, get_origin)
get_repo_group_or_error, get_perm_or_error, get_user_group_or_error,
get_origin, validate_repo_group_permissions, validate_set_owner_permissions)
from rhodecode.lib.auth import (
HasPermissionAnyApi, HasRepoGroupPermissionAnyApi,
HasUserGroupPermissionAnyApi)
from rhodecode.model.db import Session, RepoGroup
HasRepoGroupPermissionAnyApi, HasUserGroupPermissionAnyApi)
from rhodecode.model.db import Session
from rhodecode.model.repo_group import RepoGroupModel
from rhodecode.model.scm import RepoGroupList
from rhodecode.model import validation_schema
from rhodecode.model.validation_schema.schemas import repo_group_schema
@ -142,21 +141,24 @@ def get_repo_groups(request, apiuser):
@jsonrpc_method()
def create_repo_group(request, apiuser, group_name, description=Optional(''),
owner=Optional(OAttr('apiuser')),
copy_permissions=Optional(False)):
def create_repo_group(
request, apiuser, group_name,
owner=Optional(OAttr('apiuser')),
description=Optional(''),
copy_permissions=Optional(False)):
"""
Creates a repository group.
* If the repository group name contains "/", all the required repository
groups will be created.
* If the repository group name contains "/", repository group will be
created inside a repository group or nested repository groups
For example "foo/bar/baz" will create |repo| groups "foo" and "bar"
(with "foo" as parent). It will also create the "baz" repository
with "bar" as |repo| group.
For example "foo/bar/group1" will create repository group called "group1"
inside group "foo/bar". You have to have permissions to access and
write to the last repository group ("bar" in this example)
This command can only be run using an |authtoken| with admin
permissions.
This command can only be run using an |authtoken| with at least
permissions to create repository groups, or admin permissions to
parent repository groups.
:param apiuser: This is filled automatically from the |authtoken|.
:type apiuser: AuthUser
@ -193,72 +195,64 @@ def create_repo_group(request, apiuser, group_name, description=Optional(''),
"""
schema = repo_group_schema.RepoGroupSchema()
try:
data = schema.deserialize({
'group_name': group_name
})
except colander.Invalid as e:
raise JSONRPCError("Validation failed: %s" % (e.asdict(),))
group_name = data['group_name']
owner = validate_set_owner_permissions(apiuser, owner)
if isinstance(owner, Optional):
owner = apiuser.user_id
group_description = Optional.extract(description)
description = Optional.extract(description)
copy_permissions = Optional.extract(copy_permissions)
# get by full name with parents, check if it already exist
if RepoGroup.get_by_group_name(group_name):
raise JSONRPCError("repo group `%s` already exist" % (group_name,))
schema = repo_group_schema.RepoGroupSchema().bind(
# user caller
user=apiuser)
(group_name_cleaned,
parent_group_name) = RepoGroupModel()._get_group_name_and_parent(
group_name)
try:
schema_data = schema.deserialize(dict(
repo_group_name=group_name,
repo_group_owner=owner.username,
repo_group_description=description,
repo_group_copy_permissions=copy_permissions,
))
except validation_schema.Invalid as err:
raise JSONRPCValidationError(colander_exc=err)
parent_group = None
if parent_group_name:
parent_group = get_repo_group_or_error(parent_group_name)
if not HasPermissionAnyApi(
'hg.admin', 'hg.repogroup.create.true')(user=apiuser):
# check if we have admin permission for this parent repo group !
# users without admin or hg.repogroup.create can only create other
# groups in groups they own so this is a required, but can be empty
parent_group = getattr(parent_group, 'group_name', '')
_perms = ('group.admin',)
if not HasRepoGroupPermissionAnyApi(*_perms)(
user=apiuser, group_name=parent_group):
raise JSONRPCForbidden()
validated_group_name = schema_data['repo_group_name']
try:
repo_group = RepoGroupModel().create(
group_name=group_name,
group_description=group_description,
owner=owner,
copy_permissions=copy_permissions)
group_name=validated_group_name,
group_description=schema_data['repo_group_name'],
copy_permissions=schema_data['repo_group_copy_permissions'])
Session().commit()
return {
'msg': 'Created new repo group `%s`' % group_name,
'msg': 'Created new repo group `%s`' % validated_group_name,
'repo_group': repo_group.get_api_data()
}
except Exception:
log.exception("Exception occurred while trying create repo group")
raise JSONRPCError(
'failed to create repo group `%s`' % (group_name,))
'failed to create repo group `%s`' % (validated_group_name,))
@jsonrpc_method()
def update_repo_group(
request, apiuser, repogroupid, group_name=Optional(''),
description=Optional(''), owner=Optional(OAttr('apiuser')),
parent=Optional(None), enable_locking=Optional(False)):
enable_locking=Optional(False)):
"""
Updates repository group with the details given.
This command can only be run using an |authtoken| with admin
permissions.
* If the group_name name contains "/", repository group will be updated
accordingly with a repository group or nested repository groups
For example repogroupid=group-test group_name="foo/bar/group-test"
will update repository group called "group-test" and place it
inside group "foo/bar".
You have to have permissions to access and write to the last repository
group ("bar" in this example)
:param apiuser: This is filled automatically from the |authtoken|.
:type apiuser: AuthUser
:param repogroupid: Set the ID of repository group.
@ -269,29 +263,55 @@ def update_repo_group(
:type description: str
:param owner: Set the |repo| group owner.
:type owner: str
:param parent: Set the |repo| group parent.
:type parent: str or int
:param enable_locking: Enable |repo| locking. The default is false.
:type enable_locking: bool
"""
repo_group = get_repo_group_or_error(repogroupid)
if not has_superadmin_permission(apiuser):
# check if we have admin permission for this repo group !
_perms = ('group.admin',)
if not HasRepoGroupPermissionAnyApi(*_perms)(
user=apiuser, group_name=repo_group.group_name):
raise JSONRPCError(
'repository group `%s` does not exist' % (repogroupid,))
updates = {}
if not has_superadmin_permission(apiuser):
validate_repo_group_permissions(
apiuser, repogroupid, repo_group, ('group.admin',))
updates = dict(
group_name=group_name
if not isinstance(group_name, Optional) else repo_group.group_name,
group_description=description
if not isinstance(description, Optional) else repo_group.group_description,
user=owner
if not isinstance(owner, Optional) else repo_group.user.username,
enable_locking=enable_locking
if not isinstance(enable_locking, Optional) else repo_group.enable_locking
)
schema = repo_group_schema.RepoGroupSchema().bind(
# user caller
user=apiuser,
old_values=repo_group.get_api_data())
try:
store_update(updates, group_name, 'group_name')
store_update(updates, description, 'group_description')
store_update(updates, owner, 'user')
store_update(updates, parent, 'group_parent_id')
store_update(updates, enable_locking, 'enable_locking')
repo_group = RepoGroupModel().update(repo_group, updates)
schema_data = schema.deserialize(dict(
repo_group_name=updates['group_name'],
repo_group_owner=updates['user'],
repo_group_description=updates['group_description'],
repo_group_enable_locking=updates['enable_locking'],
))
except validation_schema.Invalid as err:
raise JSONRPCValidationError(colander_exc=err)
validated_updates = dict(
group_name=schema_data['repo_group']['repo_group_name_without_group'],
group_parent_id=schema_data['repo_group']['repo_group_id'],
user=schema_data['repo_group_owner'],
group_description=schema_data['repo_group_description'],
enable_locking=schema_data['repo_group_enable_locking'],
)
try:
RepoGroupModel().update(repo_group, validated_updates)
Session().commit()
return {
'msg': 'updated repository group ID:%s %s' % (
@ -299,7 +319,9 @@ def update_repo_group(
'repo_group': repo_group.get_api_data()
}
except Exception:
log.exception("Exception occurred while trying update repo group")
log.exception(
u"Exception occurred while trying update repo group %s",
repogroupid)
raise JSONRPCError('failed to update repository group `%s`'
% (repogroupid,))
@ -340,12 +362,9 @@ def delete_repo_group(request, apiuser, repogroupid):
repo_group = get_repo_group_or_error(repogroupid)
if not has_superadmin_permission(apiuser):
# check if we have admin permission for this repo group !
_perms = ('group.admin',)
if not HasRepoGroupPermissionAnyApi(*_perms)(
user=apiuser, group_name=repo_group.group_name):
raise JSONRPCError(
'repository group `%s` does not exist' % (repogroupid,))
validate_repo_group_permissions(
apiuser, repogroupid, repo_group, ('group.admin',))
try:
RepoGroupModel().delete(repo_group)
Session().commit()
@ -408,12 +427,8 @@ def grant_user_permission_to_repo_group(
repo_group = get_repo_group_or_error(repogroupid)
if not has_superadmin_permission(apiuser):
# check if we have admin permission for this repo group !
_perms = ('group.admin',)
if not HasRepoGroupPermissionAnyApi(*_perms)(
user=apiuser, group_name=repo_group.group_name):
raise JSONRPCError(
'repository group `%s` does not exist' % (repogroupid,))
validate_repo_group_permissions(
apiuser, repogroupid, repo_group, ('group.admin',))
user = get_user_or_error(userid)
perm = get_perm_or_error(perm, prefix='group.')
@ -487,12 +502,8 @@ def revoke_user_permission_from_repo_group(
repo_group = get_repo_group_or_error(repogroupid)
if not has_superadmin_permission(apiuser):
# check if we have admin permission for this repo group !
_perms = ('group.admin',)
if not HasRepoGroupPermissionAnyApi(*_perms)(
user=apiuser, group_name=repo_group.group_name):
raise JSONRPCError(
'repository group `%s` does not exist' % (repogroupid,))
validate_repo_group_permissions(
apiuser, repogroupid, repo_group, ('group.admin',))
user = get_user_or_error(userid)
apply_to_children = Optional.extract(apply_to_children)
@ -569,12 +580,8 @@ def grant_user_group_permission_to_repo_group(
perm = get_perm_or_error(perm, prefix='group.')
user_group = get_user_group_or_error(usergroupid)
if not has_superadmin_permission(apiuser):
# check if we have admin permission for this repo group !
_perms = ('group.admin',)
if not HasRepoGroupPermissionAnyApi(*_perms)(
user=apiuser, group_name=repo_group.group_name):
raise JSONRPCError(
'repository group `%s` does not exist' % (repogroupid,))
validate_repo_group_permissions(
apiuser, repogroupid, repo_group, ('group.admin',))
# check if we have at least read permission for this user group !
_perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
@ -656,12 +663,8 @@ def revoke_user_group_permission_from_repo_group(
repo_group = get_repo_group_or_error(repogroupid)
user_group = get_user_group_or_error(usergroupid)
if not has_superadmin_permission(apiuser):
# check if we have admin permission for this repo group !
_perms = ('group.admin',)
if not HasRepoGroupPermissionAnyApi(*_perms)(
user=apiuser, group_name=repo_group.group_name):
raise JSONRPCError(
'repository group `%s` does not exist' % (repogroupid,))
validate_repo_group_permissions(
apiuser, repogroupid, repo_group, ('group.admin',))
# check if we have at least read permission for this user group !
_perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)

215
rhodecode/model/validation_schema/schemas/repo_group_schema.py
View file

@ -21,9 +21,220 @@
import colander
from rhodecode.translation import _
from rhodecode.model.validation_schema import validators, preparers, types
def get_group_and_repo(repo_name):
from rhodecode.model.repo_group import RepoGroupModel
return RepoGroupModel()._get_group_name_and_parent(
repo_name, get_object=True)
@colander.deferred
def deferred_can_write_to_group_validator(node, kw):
old_values = kw.get('old_values') or {}
request_user = kw.get('user')
def can_write_group_validator(node, value):
from rhodecode.lib.auth import (
HasPermissionAny, HasRepoGroupPermissionAny)
from rhodecode.model.repo_group import RepoGroupModel
messages = {
'invalid_parent_repo_group':
_(u"Parent repository group `{}` does not exist"),
# permissions denied we expose as not existing, to prevent
# resource discovery
'permission_denied_parent_group':
_(u"Parent repository group `{}` does not exist"),
'permission_denied_root':
_(u"You do not have the permission to store "
u"repository groups in the root location.")
}
value = value['repo_group_name']
parent_group_name = value
is_root_location = value is types.RootLocation
# NOT initialized validators, we must call them
can_create_repo_groups_at_root = HasPermissionAny(
'hg.admin', 'hg.repogroup.create.true')
if is_root_location:
if can_create_repo_groups_at_root(user=request_user):
# we can create repo group inside tool-level. No more checks
# are required
return
else:
raise colander.Invalid(node, messages['permission_denied_root'])
# check if the parent repo group actually exists
parent_group = None
if parent_group_name:
parent_group = RepoGroupModel().get_by_group_name(parent_group_name)
if value and not parent_group:
raise colander.Invalid(
node, messages['invalid_parent_repo_group'].format(
parent_group_name))
# check if we have permissions to create new groups under
# parent repo group
# create repositories with write permission on group is set to true
create_on_write = HasPermissionAny(
'hg.create.write_on_repogroup.true')(user=request_user)
group_admin = HasRepoGroupPermissionAny('group.admin')(
parent_group_name, 'can write into group validator', user=request_user)
group_write = HasRepoGroupPermissionAny('group.write')(
parent_group_name, 'can write into group validator', user=request_user)
# creation by write access is currently disabled. Needs thinking if
# we want to allow this...
forbidden = not (group_admin or (group_write and create_on_write and 0))
if parent_group and forbidden:
msg = messages['permission_denied_parent_group'].format(
parent_group_name)
raise colander.Invalid(node, msg)
return can_write_group_validator
@colander.deferred
def deferred_repo_group_owner_validator(node, kw):
def repo_owner_validator(node, value):
from rhodecode.model.db import User
existing = User.get_by_username(value)
if not existing:
msg = _(u'Repo group owner with id `{}` does not exists').format(
value)
raise colander.Invalid(node, msg)
return repo_owner_validator
@colander.deferred
def deferred_unique_name_validator(node, kw):
request_user = kw.get('user')
old_values = kw.get('old_values') or {}
def unique_name_validator(node, value):
from rhodecode.model.db import Repository, RepoGroup
name_changed = value != old_values.get('group_name')
existing = Repository.get_by_repo_name(value)
if name_changed and existing:
msg = _(u'Repository with name `{}` already exists').format(value)
raise colander.Invalid(node, msg)
existing_group = RepoGroup.get_by_group_name(value)
if name_changed and existing_group:
msg = _(u'Repository group with name `{}` already exists').format(
value)
raise colander.Invalid(node, msg)
return unique_name_validator
@colander.deferred
def deferred_repo_group_name_validator(node, kw):
return validators.valid_name_validator
class GroupType(colander.Mapping):
def _validate(self, node, value):
try:
return dict(repo_group_name=value)
except Exception as e:
raise colander.Invalid(
node, '"${val}" is not a mapping type: ${err}'.format(
val=value, err=e))
def deserialize(self, node, cstruct):
if cstruct is colander.null:
return cstruct
appstruct = super(GroupType, self).deserialize(node, cstruct)
validated_name = appstruct['repo_group_name']
# inject group based on once deserialized data
(repo_group_name_without_group,
parent_group_name,
parent_group) = get_group_and_repo(validated_name)
appstruct['repo_group_name_without_group'] = repo_group_name_without_group
appstruct['repo_group_name'] = parent_group_name or types.RootLocation
if parent_group:
appstruct['repo_group_id'] = parent_group.group_id
return appstruct
class GroupSchema(colander.SchemaNode):
schema_type = GroupType
validator = deferred_can_write_to_group_validator
missing = colander.null
class RepoGroup(GroupSchema):
repo_group_name = colander.SchemaNode(
types.GroupNameType())
repo_group_id = colander.SchemaNode(
colander.String(), missing=None)
repo_group_name_without_group = colander.SchemaNode(
colander.String(), missing=None)
class RepoGroupAccessSchema(colander.MappingSchema):
repo_group = RepoGroup()
class RepoGroupNameUniqueSchema(colander.MappingSchema):
unique_repo_group_name = colander.SchemaNode(
colander.String(),
validator=deferred_unique_name_validator)
class RepoGroupSchema(colander.Schema):
group_name = colander.SchemaNode(types.GroupNameType())
repo_group_name = colander.SchemaNode(
types.GroupNameType(),
validator=deferred_repo_group_name_validator)
repo_group_owner = colander.SchemaNode(
colander.String(),
validator=deferred_repo_group_owner_validator)
repo_group_description = colander.SchemaNode(
colander.String(), missing='')
repo_group_copy_permissions = colander.SchemaNode(
types.StringBooleanType(),
missing=False)
repo_group_enable_locking = colander.SchemaNode(
types.StringBooleanType(),
missing=False)
def deserialize(self, cstruct):
"""
Custom deserialize that allows to chain validation, and verify
permissions, and as last step uniqueness
"""
appstruct = super(RepoGroupSchema, self).deserialize(cstruct)
validated_name = appstruct['repo_group_name']
# second pass to validate permissions to repo_group
second = RepoGroupAccessSchema().bind(**self.bindings)
appstruct_second = second.deserialize({'repo_group': validated_name})
# save result
appstruct['repo_group'] = appstruct_second['repo_group']
# thirds to validate uniqueness
third = RepoGroupNameUniqueSchema().bind(**self.bindings)
third.deserialize({'unique_repo_group_name': validated_name})
return appstruct

116
rhodecode/tests/models/schemas/test_repo_group_schema.py Normal file
View file

@ -0,0 +1,116 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016-2016 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import 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_owner=user_admin.username
))
assert schema_data['repo_group_name'] == 'dupa'
assert schema_data['repo_group'] == {
'repo_group_id': None,
'repo_group_name': types.RootLocation,
'repo_group_name_without_group': 'dupa'}
@pytest.mark.parametrize('given, err_key, expected_exc', [
('xxx/dupa', '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'
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'}
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'
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