repo-group-schemas: refactor repository group schemas.

- fixes  #4133
- refactor to make it behave like repo schemas
- chain validation fixes
- nicer error messages
- consistent permission checks

rhodecode/tests/models/schemas/test_repo_group_schema.py

@@ -0,0 +1,116 b''
+# -*- 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

rhodecode/api/tests/test_create_repo_group.py

@@ -54,55 +54,10 b' 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 b' 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 b' 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 b' 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)
-        fixture.destroy_repo_group(repo_group_name)
+
+    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)
 
     @mock.patch.object(RepoGroupModel, 'create', crash)
     def test_api_create_repo_group_exception_occurred(self):

rhodecode/api/tests/test_update_repo_group.py

@@ -30,22 +30,30 b' 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_parent(self, user_util):
+    def test_update_group_name_change_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, initial_name)
+        expected_group_name = '{}/{}'.format(parent_group_name, 'new-group')
+        initial_name = self._update(user_util, group_name=expected_group_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 b' 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 = {
+            '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_name,)
+        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 b' 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 b' 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

rhodecode/api/views/repo_group_api.py

@@ -21,19 +21,18 b''
 
 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 b' 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 b' def create_repo_group(request, apiuser, '
 
     """
 
-    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,))
-
-    (group_name_cleaned,
-     parent_group_name) = RepoGroupModel()._get_group_name_and_parent(
-        group_name)
+    schema = repo_group_schema.RepoGroupSchema().bind(
+        # user caller
+        user=apiuser)
 
-    parent_group = None
-    if parent_group_name:
-        parent_group = get_repo_group_or_error(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)
 
-    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 b' 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,))
+        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
+    )
 
-    updates = {}
+    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 b' 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 b' def delete_repo_group(request, apiuser, '
 
     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 b' 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 b' def revoke_user_permission_from_repo_gro'
     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 b' def grant_user_group_permission_to_repo_'
     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 b' def revoke_user_group_permission_from_re'
     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',)

rhodecode/model/validation_schema/schemas/repo_group_schema.py

@@ -21,9 +21,220 b''
 
 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)
+
+
[e-mail]
+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
+
+
[e-mail]
+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
+
+
[e-mail]
+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
+
+
[e-mail]
+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