repo-groups: fixed a regression for updating nested repository groups

rhodecode/apps/repo_group/views/repo_group_settings.py

@@ -1,183 +1,183 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2011-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 logging
 import deform
 
 from pyramid.view import view_config
 from pyramid.httpexceptions import HTTPFound
 
 from rhodecode.apps._base import RepoGroupAppView
 from rhodecode.forms import RcForm
 from rhodecode.lib import helpers as h
 from rhodecode.lib import audit_logger
 from rhodecode.lib.auth import (
     LoginRequired, HasPermissionAll,
     HasRepoGroupPermissionAny, HasRepoGroupPermissionAnyDecorator, CSRFRequired)
 from rhodecode.model.db import Session, RepoGroup
 from rhodecode.model.scm import RepoGroupList
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.validation_schema.schemas import repo_group_schema
 
 log = logging.getLogger(__name__)
 
 
 class RepoGroupSettingsView(RepoGroupAppView):
     def load_default_context(self):
         c = self._get_local_tmpl_context()
         c.repo_group = self.db_repo_group
         no_parrent = not c.repo_group.parent_group
         can_create_in_root = self._can_create_repo_group()
 
         show_root_location = False
         if no_parrent or can_create_in_root:
             # we're global admin, we're ok and we can create TOP level groups
             # or in case this group is already at top-level we also allow
             # creation in root
             show_root_location = True
 
         acl_groups = RepoGroupList(
             RepoGroup.query().all(),
             perm_set=['group.admin'])
         c.repo_groups = RepoGroup.groups_choices(
             groups=acl_groups,
             show_empty_group=show_root_location)
         # filter out current repo group
         exclude_group_ids = [c.repo_group.group_id]
         c.repo_groups = filter(lambda x: x[0] not in exclude_group_ids,
                                c.repo_groups)
         c.repo_groups_choices = map(lambda k: k[0], c.repo_groups)
 
         parent_group = c.repo_group.parent_group
 
         add_parent_group = (parent_group and (
             parent_group.group_id not in c.repo_groups_choices))
         if add_parent_group:
             c.repo_groups_choices.append(parent_group.group_id)
             c.repo_groups.append(RepoGroup._generate_choice(parent_group))
 
         self._register_global_c(c)
         return c
 
     def _can_create_repo_group(self, parent_group_id=None):
         is_admin = HasPermissionAll('hg.admin')('group create controller')
         create_repo_group = HasPermissionAll(
             'hg.repogroup.create.true')('group create controller')
         if is_admin or (create_repo_group and not parent_group_id):
             # we're global admin, or we have global repo group create
             # permission
             # we're ok and we can create TOP level groups
             return True
         elif parent_group_id:
             # we check the permission if we can write to parent group
             group = RepoGroup.get(parent_group_id)
             group_name = group.group_name if group else None
             if HasRepoGroupPermissionAny('group.admin')(
                     group_name, 'check if user is an admin of group'):
                 # we're an admin of passed in group, we're ok.
                 return True
             else:
                 return False
         return False
 
     def _get_schema(self, c, old_values=None):
         return repo_group_schema.RepoGroupSettingsSchema().bind(
             repo_group_repo_group_options=c.repo_groups_choices,
             repo_group_repo_group_items=c.repo_groups,
 
             # user caller
             user=self._rhodecode_user,
             old_values=old_values
         )
 
     @LoginRequired()
     @HasRepoGroupPermissionAnyDecorator('group.admin')
     @view_config(
         route_name='edit_repo_group', request_method='GET',
         renderer='rhodecode:templates/admin/repo_groups/repo_group_edit.mako')
     def edit_settings(self):
         c = self.load_default_context()
         c.active = 'settings'
 
         defaults = RepoGroupModel()._get_defaults(self.db_repo_group_name)
         defaults['repo_group_owner'] = defaults['user']
 
         schema = self._get_schema(c)
         c.form = RcForm(schema, appstruct=defaults)
         return self._get_template_context(c)
 
     @LoginRequired()
     @HasRepoGroupPermissionAnyDecorator('group.admin')
     @CSRFRequired()
     @view_config(
         route_name='edit_repo_group', request_method='POST',
         renderer='rhodecode:templates/admin/repo_groups/repo_group_edit.mako')
     def edit_settings_update(self):
         _ = self.request.translate
         c = self.load_default_context()
         c.active = 'settings'
 
         old_repo_group_name = self.db_repo_group_name
         new_repo_group_name = old_repo_group_name
 
         old_values = RepoGroupModel()._get_defaults(self.db_repo_group_name)
         schema = self._get_schema(c, old_values=old_values)
 
         c.form = RcForm(schema)
         pstruct = self.request.POST.items()
 
         try:
             schema_data = c.form.validate(pstruct)
         except deform.ValidationFailure as err_form:
             return self._get_template_context(c)
 
         # data is now VALID, proceed with updates
         # save validated data back into the updates dict
         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(self.db_repo_group, validated_updates)
 
             audit_logger.store_web(
                 'repo_group.edit', action_data={'old_data': old_values},
                 user=c.rhodecode_user)
 
             Session().commit()
 
             # use the new full name for redirect once we know we updated
             # the name on filesystem and in DB
-            new_repo_group_name = schema_data['repo_group_name']
+            new_repo_group_name = schema_data['repo_group']['repo_group_name_with_group']
 
             h.flash(_('Repository Group `{}` updated successfully').format(
                 old_repo_group_name), category='success')
 
         except Exception:
             log.exception("Exception during update or repository group")
             h.flash(_('Error occurred during update of repository group %s')
                     % old_repo_group_name, category='error')
 
         raise HTTPFound(
             h.route_path('edit_repo_group', repo_group_name=new_repo_group_name))

rhodecode/model/validation_schema/schemas/repo_group_schema.py

@@ -1,284 +1,298 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
 import colander
 import deform.widget
 
 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)
 
 
+def get_repo_group(repo_group_id):
+    from rhodecode.model.repo_group import RepoGroup
+    return RepoGroup.get(repo_group_id), RepoGroup.CHOICES_SEPARATOR
+
+
 @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
 
 
 @colander.deferred
 def deferred_repo_group_validator(node, kw):
     options = kw.get(
         'repo_group_repo_group_options')
     return colander.OneOf([x for x in options])
 
 
 @colander.deferred
 def deferred_repo_group_widget(node, kw):
     items = kw.get('repo_group_repo_group_items')
     return deform.widget.Select2Widget(values=items)
 
 
 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_with_group'] = 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):
 
     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='', widget=deform.widget.TextAreaWidget())
 
     repo_group_copy_permissions = colander.SchemaNode(
         types.StringBooleanType(),
         missing=False, widget=deform.widget.CheckboxWidget())
 
     repo_group_enable_locking = colander.SchemaNode(
         types.StringBooleanType(),
         missing=False, widget=deform.widget.CheckboxWidget())
 
     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
 
 
 class RepoGroupSettingsSchema(RepoGroupSchema):
     repo_group = colander.SchemaNode(
         colander.Integer(),
         validator=deferred_repo_group_validator,
         widget=deferred_repo_group_widget,
         missing='')
 
     def deserialize(self, cstruct):
         """
         Custom deserialize that allows to chain validation, and verify
         permissions, and as last step uniqueness
         """
 
         # first pass, to validate given data
         appstruct = super(RepoGroupSchema, self).deserialize(cstruct)
         validated_name = appstruct['repo_group_name']
 
+        # because of repoSchema adds repo-group as an ID, we inject it as
+        # full name here because validators require it, it's unwrapped later
+        # so it's safe to use and final name is going to be without group anyway
+
+        group, separator = get_repo_group(appstruct['repo_group'])
+        if group:
+            validated_name = separator.join([group.group_name, validated_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

rhodecode/tests/models/schemas/test_repo_group_schema.py

@@ -1,116 +1,118 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import colander
 import pytest
 
 from rhodecode.model.validation_schema import types
 from rhodecode.model.validation_schema.schemas import repo_group_schema
 
 
 class TestRepoGroupSchema(object):
 
     @pytest.mark.parametrize('given, expected', [
         ('my repo', 'my-repo'),
         (' hello   world mike ', 'hello-world-mike'),
 
         ('//group1/group2//', 'group1/group2'),
         ('//group1///group2//', 'group1/group2'),
         ('///group1/group2///group3', 'group1/group2/group3'),
         ('word g1/group2///group3', 'word-g1/group2/group3'),
 
         ('grou p1/gro;,,##up2//.../group3', 'grou-p1/group2/group3'),
 
         ('group,,,/,,,/1/2/3', 'group/1/2/3'),
         ('grou[]p1/gro;up2///gro up3', 'group1/group2/gro-up3'),
         (u'grou[]p1/gro;up2///gro up3/ąć', u'group1/group2/gro-up3/ąć'),
     ])
     def test_deserialize_repo_name(self, app, user_admin, given, expected):
         schema = repo_group_schema.RepoGroupSchema().bind()
         assert schema.get('repo_group_name').deserialize(given) == expected
 
     def test_deserialize(self, app, user_admin):
         schema = repo_group_schema.RepoGroupSchema().bind(
             user=user_admin
         )
 
         schema_data = schema.deserialize(dict(
             repo_group_name='my_schema_group',
             repo_group_owner=user_admin.username
         ))
 
         assert schema_data['repo_group_name'] == u'my_schema_group'
         assert schema_data['repo_group'] == {
             'repo_group_id': None,
             'repo_group_name': types.RootLocation,
+            'repo_group_name_with_group': u'my_schema_group',
             'repo_group_name_without_group': u'my_schema_group'}
 
     @pytest.mark.parametrize('given, err_key, expected_exc', [
         ('xxx/my_schema_group', 'repo_group', 'Parent repository group `xxx` does not exist'),
         ('', 'repo_group_name', 'Name must start with a letter or number. Got ``'),
     ])
     def test_deserialize_with_bad_group_name(
             self, app, user_admin, given, err_key, expected_exc):
         schema = repo_group_schema.RepoGroupSchema().bind(
             repo_type_options=['hg'],
             user=user_admin
         )
 
         with pytest.raises(colander.Invalid) as excinfo:
             schema.deserialize(dict(
                 repo_group_name=given,
                 repo_group_owner=user_admin.username
             ))
 
         assert excinfo.value.asdict()[err_key] == expected_exc
 
     def test_deserialize_with_group_name(self, app, user_admin, test_repo_group):
         schema = repo_group_schema.RepoGroupSchema().bind(
             user=user_admin
         )
 
         full_name = test_repo_group.group_name + u'/my_schema_group'
         schema_data = schema.deserialize(dict(
             repo_group_name=full_name,
             repo_group_owner=user_admin.username
         ))
 
         assert schema_data['repo_group_name'] == full_name
         assert schema_data['repo_group'] == {
             'repo_group_id': test_repo_group.group_id,
             'repo_group_name': test_repo_group.group_name,
+            'repo_group_name_with_group': full_name,
             'repo_group_name_without_group': u'my_schema_group'}
 
     def test_deserialize_with_group_name_regular_user_no_perms(
             self, app, user_regular, test_repo_group):
         schema = repo_group_schema.RepoGroupSchema().bind(
             user=user_regular
         )
 
         full_name = test_repo_group.group_name + u'/my_schema_group'
         with pytest.raises(colander.Invalid) as excinfo:
             schema.deserialize(dict(
                 repo_group_name=full_name,
                 repo_group_owner=user_regular.username
             ))
 
         expected = 'Parent repository group `{}` does not exist'.format(
             test_repo_group.group_name)
         assert excinfo.value.asdict()['repo_group'] == expected