Changeset - 4707d09d49a6
[Not reviewed]
default
0 2 0
Mads Kiilerich - 7 years ago 2018-06-04 14:26:37
mads@kiilerich.com
repos: make edit_permissions_revoke controller return an actual (empty) body on success

The code 204 was slightly confusing even though technically OK.
2 files changed with 2 insertions and 1 deletions:
0 comments (0 inline, 0 general)
kallithea/controllers/admin/repos.py
Show inline comments
 
@@ -262,192 +262,193 @@ class ReposController(BaseRepoController
 
        repo_model = RepoModel()
 
        repo = repo_model.get_by_repo_name(repo_name)
 
        if not repo:
 
            h.not_mapped_error(repo_name)
 
            raise HTTPFound(location=url('repos'))
 
        try:
 
            _forks = repo.forks.count()
 
            handle_forks = None
 
            if _forks and request.POST.get('forks'):
 
                do = request.POST['forks']
 
                if do == 'detach_forks':
 
                    handle_forks = 'detach'
 
                    h.flash(_('Detached %s forks') % _forks, category='success')
 
                elif do == 'delete_forks':
 
                    handle_forks = 'delete'
 
                    h.flash(_('Deleted %s forks') % _forks, category='success')
 
            repo_model.delete(repo, forks=handle_forks)
 
            action_logger(request.authuser, 'admin_deleted_repo',
 
                repo_name, request.ip_addr)
 
            ScmModel().mark_for_invalidation(repo_name)
 
            h.flash(_('Deleted repository %s') % repo_name, category='success')
 
            Session().commit()
 
        except AttachedForksError:
 
            h.flash(_('Cannot delete repository %s which still has forks')
 
                        % repo_name, category='warning')
 

	
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            h.flash(_('An error occurred during deletion of %s') % repo_name,
 
                    category='error')
 

	
 
        if repo.group:
 
            raise HTTPFound(location=url('repos_group_home', group_name=repo.group.group_name))
 
        raise HTTPFound(location=url('repos'))
 

	
 
    @HasRepoPermissionLevelDecorator('admin')
 
    def edit(self, repo_name):
 
        defaults = self.__load_data()
 
        c.repo_fields = RepositoryField.query() \
 
            .filter(RepositoryField.repository == c.repo_info).all()
 
        c.active = 'settings'
 
        return htmlfill.render(
 
            render('admin/repos/repo_edit.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasRepoPermissionLevelDecorator('admin')
 
    def edit_permissions(self, repo_name):
 
        c.repo_info = self._load_repo()
 
        c.active = 'permissions'
 
        defaults = RepoModel()._get_defaults(repo_name)
 

	
 
        return htmlfill.render(
 
            render('admin/repos/repo_edit.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasRepoPermissionLevelDecorator('admin')
 
    def edit_permissions_update(self, repo_name):
 
        form = RepoPermsForm()().to_python(request.POST)
 
        RepoModel()._update_permissions(repo_name, form['perms_new'],
 
                                        form['perms_updates'])
 
        # TODO: implement this
 
        #action_logger(request.authuser, 'admin_changed_repo_permissions',
 
        #              repo_name, request.ip_addr)
 
        Session().commit()
 
        h.flash(_('Repository permissions updated'), category='success')
 
        raise HTTPFound(location=url('edit_repo_perms', repo_name=repo_name))
 

	
 
    @HasRepoPermissionLevelDecorator('admin')
 
    def edit_permissions_revoke(self, repo_name):
 
        try:
 
            obj_type = request.POST.get('obj_type')
 
            obj_id = None
 
            if obj_type == 'user':
 
                obj_id = safe_int(request.POST.get('user_id'))
 
            elif obj_type == 'user_group':
 
                obj_id = safe_int(request.POST.get('user_group_id'))
 

	
 
            if obj_type == 'user':
 
                RepoModel().revoke_user_permission(repo=repo_name, user=obj_id)
 
            elif obj_type == 'user_group':
 
                RepoModel().revoke_user_group_permission(
 
                    repo=repo_name, group_name=obj_id
 
                )
 
            # TODO: implement this
 
            #action_logger(request.authuser, 'admin_revoked_repo_permissions',
 
            #              repo_name, request.ip_addr)
 
            Session().commit()
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            h.flash(_('An error occurred during revoking of permission'),
 
                    category='error')
 
            raise HTTPInternalServerError()
 
        return []
 

	
 
    @HasRepoPermissionLevelDecorator('admin')
 
    def edit_fields(self, repo_name):
 
        c.repo_info = self._load_repo()
 
        c.repo_fields = RepositoryField.query() \
 
            .filter(RepositoryField.repository == c.repo_info).all()
 
        c.active = 'fields'
 
        if request.POST:
 

	
 
            raise HTTPFound(location=url('repo_edit_fields'))
 
        return render('admin/repos/repo_edit.html')
 

	
 
    @HasRepoPermissionLevelDecorator('admin')
 
    def create_repo_field(self, repo_name):
 
        try:
 
            form_result = RepoFieldForm()().to_python(dict(request.POST))
 
            new_field = RepositoryField()
 
            new_field.repository = Repository.get_by_repo_name(repo_name)
 
            new_field.field_key = form_result['new_field_key']
 
            new_field.field_type = form_result['new_field_type']  # python type
 
            new_field.field_value = form_result['new_field_value']  # set initial blank value
 
            new_field.field_desc = form_result['new_field_desc']
 
            new_field.field_label = form_result['new_field_label']
 
            Session().add(new_field)
 
            Session().commit()
 
        except formencode.Invalid as e:
 
            h.flash(_('Field validation error: %s') % e.msg, category='error')
 
        except Exception as e:
 
            log.error(traceback.format_exc())
 
            h.flash(_('An error occurred during creation of field: %r') % e, category='error')
 
        raise HTTPFound(location=url('edit_repo_fields', repo_name=repo_name))
 

	
 
    @HasRepoPermissionLevelDecorator('admin')
 
    def delete_repo_field(self, repo_name, field_id):
 
        field = RepositoryField.get_or_404(field_id)
 
        try:
 
            Session().delete(field)
 
            Session().commit()
 
        except Exception as e:
 
            log.error(traceback.format_exc())
 
            msg = _('An error occurred during removal of field')
 
            h.flash(msg, category='error')
 
        raise HTTPFound(location=url('edit_repo_fields', repo_name=repo_name))
 

	
 
    @HasRepoPermissionLevelDecorator('admin')
 
    def edit_advanced(self, repo_name):
 
        c.repo_info = self._load_repo()
 
        c.default_user_id = User.get_default_user().user_id
 
        c.in_public_journal = UserFollowing.query() \
 
            .filter(UserFollowing.user_id == c.default_user_id) \
 
            .filter(UserFollowing.follows_repository == c.repo_info).scalar()
 

	
 
        _repos = Repository.query(sorted=True).all()
 
        read_access_repos = RepoList(_repos, perm_level='read')
 
        c.repos_list = [(None, _('-- Not a fork --'))]
 
        c.repos_list += [(x.repo_id, x.repo_name)
 
                         for x in read_access_repos
 
                         if x.repo_id != c.repo_info.repo_id]
 

	
 
        defaults = {
 
            'id_fork_of': c.repo_info.fork_id if c.repo_info.fork_id else ''
 
        }
 

	
 
        c.active = 'advanced'
 
        if request.POST:
 
            raise HTTPFound(location=url('repo_edit_advanced'))
 
        return htmlfill.render(
 
            render('admin/repos/repo_edit.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasRepoPermissionLevelDecorator('admin')
 
    def edit_advanced_journal(self, repo_name):
 
        """
 
        Sets this repository to be visible in public journal,
 
        in other words asking default user to follow this repo
 

	
 
        :param repo_name:
 
        """
 

	
 
        try:
 
            repo_id = Repository.get_by_repo_name(repo_name).repo_id
 
            user_id = User.get_default_user().user_id
 
            self.scm_model.toggle_following_repo(repo_id, user_id)
 
            h.flash(_('Updated repository visibility in public journal'),
 
                    category='success')
 
            Session().commit()
 
        except Exception:
 
            h.flash(_('An error occurred during setting this'
 
                      ' repository in public journal'),
 
                    category='error')
 
        raise HTTPFound(location=url('edit_repo_advanced', repo_name=repo_name))
 

	
 
    @HasRepoPermissionLevelDecorator('admin')
 
    def edit_advanced_fork(self, repo_name):
kallithea/tests/functional/test_admin_permissions.py
Show inline comments
 
@@ -32,98 +32,98 @@ class TestAdminPermissionsController(Tes
 
        invalidate_all_caches()
 

	
 
        self.app.get(url('admin_permissions_ips'), status=302)
 

	
 
        # REMOTE_ADDR must match 127.0.0.0/24
 
        response = self.app.get(url('admin_permissions_ips'),
 
                                extra_environ={'REMOTE_ADDR': '127.0.0.1'})
 
        response.mustcontain('127.0.0.0/24')
 
        response.mustcontain('127.0.0.0 - 127.0.0.255')
 

	
 
    def test_delete_ips(self, auto_clear_ip_permissions):
 
        self.log_user()
 
        default_user_id = User.get_default_user().user_id
 

	
 
        ## first add
 
        new_ip = '127.0.0.0/24'
 
        with test_context(self.app):
 
            user_model = UserModel()
 
            ip_obj = user_model.add_extra_ip(default_user_id, new_ip)
 
            Session().commit()
 

	
 
        ## double check that add worked
 
        # IP permissions are cached, need to invalidate this cache explicitly
 
        invalidate_all_caches()
 
        self.app.get(url('admin_permissions_ips'), status=302)
 
        # REMOTE_ADDR must match 127.0.0.0/24
 
        response = self.app.get(url('admin_permissions_ips'),
 
                                extra_environ={'REMOTE_ADDR': '127.0.0.1'})
 
        response.mustcontain('127.0.0.0/24')
 
        response.mustcontain('127.0.0.0 - 127.0.0.255')
 

	
 
        ## now delete
 
        response = self.app.post(url('edit_user_ips_delete', id=default_user_id),
 
                                 params=dict(del_ip_id=ip_obj.ip_id,
 
                                             _authentication_token=self.authentication_token()),
 
                                 extra_environ={'REMOTE_ADDR': '127.0.0.1'})
 

	
 
        # IP permissions are cached, need to invalidate this cache explicitly
 
        invalidate_all_caches()
 

	
 
        response = self.app.get(url('admin_permissions_ips'))
 
        response.mustcontain('All IP addresses are allowed')
 
        response.mustcontain(no=['127.0.0.0/24'])
 
        response.mustcontain(no=['127.0.0.0 - 127.0.0.255'])
 

	
 
    def test_index_overview(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_permissions_perms'))
 
        # Test response...
 

	
 
    def test_edit_permissions_permissions(self):
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
 

	
 
        # Test unauthenticated access - it will redirect to login page
 
        response = self.app.post(
 
            url('edit_repo_perms_update', repo_name=HG_REPO),
 
            params=dict(
 
                perm_new_member_1='repository.read',
 
                perm_new_member_name_1=user.username,
 
                perm_new_member_type_1='user',
 
                _authentication_token=self.authentication_token()),
 
            status=302)
 

	
 
        assert not response.location.endswith(url('edit_repo_perms_update', repo_name=HG_REPO))
 
        assert response.location.endswith(url('login_home', came_from=url('edit_repo_perms_update', repo_name=HG_REPO)))
 

	
 
        response = self.app.post(
 
            url('edit_repo_perms_revoke', repo_name=HG_REPO),
 
            params=dict(
 
                obj_type='user',
 
                user_id=user.user_id,
 
                _authentication_token=self.authentication_token()),
 
            status=302)
 

	
 
        assert response.location.endswith(url('login_home', came_from=url('edit_repo_perms_revoke', repo_name=HG_REPO)))
 

	
 
        # Test authenticated access
 
        self.log_user()
 

	
 
        response = self.app.post(
 
            url('edit_repo_perms_update', repo_name=HG_REPO),
 
            params=dict(
 
                perm_new_member_1='repository.read',
 
                perm_new_member_name_1=user.username,
 
                perm_new_member_type_1='user',
 
                _authentication_token=self.authentication_token()),
 
            status=302)
 

	
 
        assert response.location.endswith(url('edit_repo_perms_update', repo_name=HG_REPO))
 

	
 
        response = self.app.post(
 
            url('edit_repo_perms_revoke', repo_name=HG_REPO),
 
            params=dict(
 
                obj_type='user',
 
                user_id=user.user_id,
 
                _authentication_token=self.authentication_token()),
 
            status=204) # success has no content
 
            status=200)
 
        assert not response.body
0 comments (0 inline, 0 general)