Changeset - cb362e3439dd
Parent rev.
Child rev.
[Not reviewed]
default
0 2 0
Mads Kiilerich - 10 years ago 2015-07-23 00:52:29
madski@unity3d.com
repos: refactor repo group handling - extract generic functionality that can be used consistently

By using the right repo group list in the form, it is no longer possible for
users to pick wrong repo creation locations ... and doing so anyway will be
caught earlier on with a more generic error message at the specific place.
2 files changed with 28 insertions and 42 deletions:
kallithea/controllers/admin/repos.py
27
41
kallithea/tests/functional/test_admin_repos.py
1
1
0 comments (0 inline, 0 general)
kallithea/controllers/admin/repos.py
➞
Show inline comments
 
@@ -16,118 +16,117 @@ kallithea.controllers.admin.repos
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

			




	
	
	
		
 
Repositories controller for Kallithea

			




	
	
	
		
 

			




	
	
	
		
 
This file was forked by the Kallithea project in July 2014.

			




	
	
	
		
 
Original author and date, and relevant copyright and licensing information is below:

			




	
	
	
		
 
:created_on: Apr 7, 2010

			




	
	
	
		
 
:author: marcink

			




	
	
	
		
 
:copyright: (c) 2013 RhodeCode GmbH, and others.

			




	
	
	
		
 
:license: GPLv3, see LICENSE.md for more details.

			




	
	
	
		
 
"""

			




	
	
	
		
 

			




	
	
	
		
 
import logging

			




	
	
	
		
 
import traceback

			




	
	
	
		
 
import formencode

			




	
	
	
		
 
from formencode import htmlfill

			




	
	
	
		
 
from webob.exc import HTTPInternalServerError, HTTPForbidden, HTTPNotFound

			




	
	
	
		
 
from pylons import request, tmpl_context as c, url

			




	
	
	
		
 
from pylons.controllers.util import redirect

			




	
	
	
		
 
from pylons.i18n.translation import _

			




	
	
	
		
 
from sqlalchemy.sql.expression import func

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.lib import helpers as h

			




	
	
	
		
 
from kallithea.lib.auth import LoginRequired, HasPermissionAllDecorator, \

			




	
	
	
		
 
    HasRepoPermissionAllDecorator, NotAnonymous, HasPermissionAny, \

			




	
	
	
		
 
    HasRepoPermissionAllDecorator, NotAnonymous, HasPermissionAny, HasPermissionAll, \

			




	
	
	
		
 
    HasRepoGroupPermissionAny, HasRepoPermissionAnyDecorator

			




	
	
	
		
 
from kallithea.lib.base import BaseRepoController, render

			




	
	
	
		
 
from kallithea.lib.utils import action_logger, repo_name_slug, jsonify

			




	
	
	
		
 
from kallithea.lib.vcs import RepositoryError

			




	
	
	
		
 
from kallithea.model.meta import Session

			




	
	
	
		
 
from kallithea.model.db import User, Repository, UserFollowing, RepoGroup,\

			




	
	
	
		
 
    Setting, RepositoryField

			




	
	
	
		
 
from kallithea.model.forms import RepoForm, RepoFieldForm, RepoPermsForm

			




	
	
	
		
 
from kallithea.model.scm import ScmModel, RepoGroupList, RepoList

			




	
	
	
		
 
from kallithea.model.repo import RepoModel

			




	
	
	
		
 
from kallithea.lib.compat import json

			




	
	
	
		
 
from kallithea.lib.exceptions import AttachedForksError

			




	
	
	
		
 
from kallithea.lib.utils2 import safe_int

			




	
	
	
		
 

			




	
	
	
		
 
log = logging.getLogger(__name__)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class ReposController(BaseRepoController):

			




	
	
	
		
 
    """

			




	
	
	
		
 
    REST Controller styled on the Atom Publishing Protocol"""

			




	
	
	
		
 
    # To properly map this controller, ensure your config/routing.py

			




	
	
	
		
 
    # file has a resource setup:

			




	
	
	
		
 
    #     map.resource('repo', 'repos')

			




	
	
	
		
 

			




	
	
	
		
 
    @LoginRequired()

			




	
	
	
		
 
    def __before__(self):

			




	
	
	
		
 
        super(ReposController, self).__before__()

			




	
	
	
		
 

			




	
	
	
		
 
    def _load_repo(self, repo_name):

			




	
	
	
		
 
        repo_obj = Repository.get_by_repo_name(repo_name)

			




	
	
	
		
 

			




	
	
	
		
 
        if repo_obj is None:

			




	
	
	
		
 
            h.not_mapped_error(repo_name)

			




	
	
	
		
 
            return redirect(url('repos'))

			




	
	
	
		
 

			




	
	
	
		
 
        return repo_obj

			




	
	
	
		
 

			




	
	
	
		
 
    def __load_defaults(self, repo=None):

			




	
	
	
		
 
        acl_groups = RepoGroupList(RepoGroup.query().all(),

			




	
	
	
		
 
                               perm_set=['group.write', 'group.admin'])

			




	
	
	
		
 
        c.repo_groups = RepoGroup.groups_choices(groups=acl_groups)

			




	
	
	
		
 
        c.repo_groups_choices = map(lambda k: k[0], c.repo_groups)

			




	
	
	
		
 

			




	
	
	
		
 
        # in case someone no longer have a group.write access to a repository

			




	
	
	
		
 
        # pre fill the list with this entry, we don't care if this is the same

			




	
	
	
		
 
        # but it will allow saving repo data properly.

			




	
	
	
		
 
        top_perms = ['hg.create.repository']

			




	
	
	
		
 
        repo_group_perms = ['group.admin']

			




	
	
	
		
 
        if HasPermissionAny('hg.create.write_on_repogroup.true')():

			




	
	
	
		
 
            repo_group_perms.append('group.write')

			




	
	
	
		
 
        extras = [] if repo is None else [repo.group]

			




	
	
	
		
 

			




	
	
	
		
 
        repo_group = None

			




	
	
	
		
 
        if repo:

			




	
	
	
		
 
            repo_group = repo.group

			




	
	
	
		
 
        if repo_group and repo_group.group_id not in c.repo_groups_choices:

			




	
	
	
		
 
            c.repo_groups_choices.append(repo_group.group_id)

			




	
	
	
		
 
            c.repo_groups.append(RepoGroup._generate_choice(repo_group))

			




	
	
	
		
 
        groups = RepoGroup.query().all()

			




	
	
	
		
 
        if HasPermissionAll('hg.admin')('available repo groups'):

			




	
	
	
		
 
            groups.append(None)

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            groups = list(RepoGroupList(groups, perm_set=repo_group_perms))

			




	
	
	
		
 
            if top_perms and HasPermissionAny(*top_perms)('available repo groups'):

			




	
	
	
		
 
                groups.append(None)

			




	
	
	
		
 
            for extra in extras:

			




	
	
	
		
 
                if not any(rg == extra for rg in groups):

			




	
	
	
		
 
                    groups.append(extra)

			




	
	
	
		
 
        c.repo_groups = RepoGroup.groups_choices(groups=groups, show_empty_group=False)

			




	
	
	
		
 
        c.repo_groups_choices = [rg[0] for rg in c.repo_groups]

			




	
	
	
		
 

			




	
	
	
		
 
        choices, c.landing_revs = ScmModel().get_repo_landing_revs()

			




	
	
	
		
 
        c.landing_revs_choices = choices

			




	
	
	
		
 
        c.landing_revs_choices, c.landing_revs = ScmModel().get_repo_landing_revs(repo)

			




	
	
	
		
 

			




	
	
	
		
 
    def __load_data(self, repo_name=None):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Load defaults settings for edit, and update

			




	
	
	
		
 

			




	
	
	
		
 
        :param repo_name:

			




	
	
	
		
 
        """

			




	
	
	
		
 
        c.repo_info = self._load_repo(repo_name)

			




	
	
	
		
 
        self.__load_defaults(c.repo_info)

			




	
	
	
		
 

			




	
	
	
		
 
        ##override defaults for exact repo info here git/hg etc

			




	
	
	
		
 
        choices, c.landing_revs = ScmModel().get_repo_landing_revs(c.repo_info)

			




	
	
	
		
 
        c.landing_revs_choices = choices

			




	
	
	
		
 
        defaults = RepoModel()._get_defaults(repo_name)

			




	
	
	
		
 
        defaults['clone_uri'] = c.repo_info.clone_uri_hidden # don't show password

			




	
	
	
		
 

			




	
	
	
		
 
        return defaults

			




	
	
	
		
 

			




	
	
	
		
 
    def index(self, format='html'):

			




	
	
	
		
 
        """GET /repos: All items in the collection"""

			




	
	
	
		
 
        # url('repos')

			




	
	
	
		
 
        _list = Repository.query()\

			




	
	
	
		
 
                        .order_by(func.lower(Repository.repo_name))\

			




	
	
	
		
 
                        .all()

			




	
	
	
		
 

			




	
	
	
		
 
        c.repos_list = RepoList(_list, perm_set=['repository.admin'])

			




	
	
	
		
 
        repos_data = RepoModel().get_repos_as_dict(repos_list=c.repos_list,

			




	
	
	
		
 
                                                   admin=True,

			




	
	
	
		
 
                                                   super_user_actions=True)

			




	
	
	
		
 
        #json used to render the grid

			




	
	
	
		
 
        c.data = json.dumps(repos_data)

			




	
	
	
		
 

			




	
	
	
		
 
        return render('admin/repos/repos.html')

			




	
	
	
		
 

			




	
	
	
		
 
    @NotAnonymous()

			




	
	
	
		
 
    def create(self):

			




	
	
	
		
 
        """

			




	
	
		
 
@@ -152,70 +151,60 @@ class ReposController(BaseRepoController

			




	
	
	
		
 
        except formencode.Invalid, errors:

			




	
	
	
		
 
            log.info(errors)

			




	
	
	
		
 
            return htmlfill.render(

			




	
	
	
		
 
                render('admin/repos/repo_add.html'),

			




	
	
	
		
 
                defaults=errors.value,

			




	
	
	
		
 
                errors=errors.error_dict or {},

			




	
	
	
		
 
                prefix_error=False,

			




	
	
	
		
 
                force_defaults=False,

			




	
	
	
		
 
                encoding="UTF-8")

			




	
	
	
		
 

			




	
	
	
		
 
        except Exception:

			




	
	
	
		
 
            log.error(traceback.format_exc())

			




	
	
	
		
 
            msg = (_('Error creating repository %s')

			




	
	
	
		
 
                   % form_result.get('repo_name'))

			




	
	
	
		
 
            h.flash(msg, category='error')

			




	
	
	
		
 
            return redirect(url('home'))

			




	
	
	
		
 

			




	
	
	
		
 
        return redirect(h.url('repo_creating_home',

			




	
	
	
		
 
                              repo_name=form_result['repo_name_full'],

			




	
	
	
		
 
                              task_id=task_id))

			




	
	
	
		
 

			




	
	
	
		
 
    @NotAnonymous()

			




	
	
	
		
 
    def create_repository(self):

			




	
	
	
		
 
        """GET /_admin/create_repository: Form to create a new item"""

			




	
	
	
		
 
        self.__load_defaults()

			




	
	
	
		
 
        if not c.repo_groups:

			




	
	
	
		
 
            raise HTTPForbidden

			




	
	
	
		
 
        parent_group = request.GET.get('parent_group')

			




	
	
	
		
 
        if not HasPermissionAny('hg.admin', 'hg.create.repository')():

			




	
	
	
		
 
            #you're not super admin nor have global create permissions,

			




	
	
	
		
 
            #but maybe you have at least write permission to a parent group ?

			




	
	
	
		
 
            _gr = RepoGroup.get(parent_group)

			




	
	
	
		
 
            gr_name = _gr.group_name if _gr else None

			




	
	
	
		
 
            # create repositories with write permission on group is set to true

			




	
	
	
		
 
            create_on_write = HasPermissionAny('hg.create.write_on_repogroup.true')()

			




	
	
	
		
 
            group_admin = HasRepoGroupPermissionAny('group.admin')(group_name=gr_name)

			




	
	
	
		
 
            group_write = HasRepoGroupPermissionAny('group.write')(group_name=gr_name)

			




	
	
	
		
 
            if not (group_admin or (group_write and create_on_write)):

			




	
	
	
		
 
                raise HTTPForbidden

			




	
	
	
		
 

			




	
	
	
		
 
        acl_groups = RepoGroupList(RepoGroup.query().all(),

			




	
	
	
		
 
                               perm_set=['group.write', 'group.admin'])

			




	
	
	
		
 
        c.repo_groups = RepoGroup.groups_choices(groups=acl_groups)

			




	
	
	
		
 
        c.repo_groups_choices = map(lambda k: k[0], c.repo_groups)

			




	
	
	
		
 
        choices, c.landing_revs = ScmModel().get_repo_landing_revs()

			




	
	
	
		
 

			




	
	
	
		
 
        ## apply the defaults from defaults page

			




	
	
	
		
 
        defaults = Setting.get_default_repo_settings(strip_prefix=True)

			




	
	
	
		
 
        if parent_group:

			




	
	
	
		
 
            prg = RepoGroup.get(parent_group)

			




	
	
	
		
 
            if prg is None or not any(rgc[0] == prg.group_id

			




	
	
	
		
 
                                      for rgc in c.repo_groups):

			




	
	
	
		
 
                raise HTTPForbidden

			




	
	
	
		
 
            defaults.update({'repo_group': parent_group})

			




	
	
	
		
 

			




	
	
	
		
 
        return htmlfill.render(

			




	
	
	
		
 
            render('admin/repos/repo_add.html'),

			




	
	
	
		
 
            defaults=defaults,

			




	
	
	
		
 
            errors={},

			




	
	
	
		
 
            prefix_error=False,

			




	
	
	
		
 
            encoding="UTF-8",

			




	
	
	
		
 
            force_defaults=False)

			




	
	
	
		
 

			




	
	
	
		
 
    @LoginRequired()

			




	
	
	
		
 
    @NotAnonymous()

			




	
	
	
		
 
    def repo_creating(self, repo_name):

			




	
	
	
		
 
        c.repo = repo_name

			




	
	
	
		
 
        c.task_id = request.GET.get('task_id')

			




	
	
	
		
 
        if not c.repo:

			




	
	
	
		
 
            raise HTTPNotFound()

			




	
	
	
		
 
        return render('admin/repos/repo_creating.html')

			




	
	
	
		
 

			




	
	
	
		
 
    @LoginRequired()

			




	
	
	
		
 
    @NotAnonymous()

			




	
	
	
		
 
    @jsonify

			




	
	
	
		
 
    def repo_check(self, repo_name):

			




	
	
	
		
 
        c.repo = repo_name

			




	
	
		
 
@@ -239,58 +228,55 @@ class ReposController(BaseRepoController

			




	
	
	
		
 
                                     h.url('summary_home',

			




	
	
	
		
 
                                           repo_name=repo.repo_name))

			




	
	
	
		
 
                fork = repo.fork

			




	
	
	
		
 
                if fork:

			




	
	
	
		
 
                    fork_name = fork.repo_name

			




	
	
	
		
 
                    h.flash(h.literal(_('Forked repository %s as %s')

			




	
	
	
		
 
                            % (fork_name, repo_url)), category='success')

			




	
	
	
		
 
                else:

			




	
	
	
		
 
                    h.flash(h.literal(_('Created repository %s') % repo_url),

			




	
	
	
		
 
                            category='success')

			




	
	
	
		
 
            return {'result': True}

			




	
	
	
		
 
        return {'result': False}

			




	
	
	
		
 

			




	
	
	
		
 
    @HasRepoPermissionAllDecorator('repository.admin')

			




	
	
	
		
 
    def update(self, repo_name):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        PUT /repos/repo_name: Update an existing item"""

			




	
	
	
		
 
        # Forms posted to this method should contain a hidden field:

			




	
	
	
		
 
        #    <input type="hidden" name="_method" value="PUT" />

			




	
	
	
		
 
        # Or using helpers:

			




	
	
	
		
 
        #    h.form(url('repo', repo_name=ID),

			




	
	
	
		
 
        #           method='put')

			




	
	
	
		
 
        # url('repo', repo_name=ID)

			




	
	
	
		
 
        c.repo_info = self._load_repo(repo_name)

			




	
	
	
		
 
        self.__load_defaults(c.repo_info)

			




	
	
	
		
 
        c.active = 'settings'

			




	
	
	
		
 
        c.repo_fields = RepositoryField.query()\

			




	
	
	
		
 
            .filter(RepositoryField.repository == c.repo_info).all()

			




	
	
	
		
 
        self.__load_defaults(c.repo_info)

			




	
	
	
		
 

			




	
	
	
		
 
        repo_model = RepoModel()

			




	
	
	
		
 
        changed_name = repo_name

			




	
	
	
		
 
        #override the choices with extracted revisions !

			




	
	
	
		
 
        choices, c.landing_revs = ScmModel().get_repo_landing_revs(repo_name)

			




	
	
	
		
 
        c.landing_revs_choices = choices

			




	
	
	
		
 
        repo = Repository.get_by_repo_name(repo_name)

			




	
	
	
		
 
        old_data = {

			




	
	
	
		
 
            'repo_name': repo_name,

			




	
	
	
		
 
            'repo_group': repo.group.get_dict() if repo.group else {},

			




	
	
	
		
 
            'repo_type': repo.repo_type,

			




	
	
	
		
 
        }

			




	
	
	
		
 
        _form = RepoForm(edit=True, old_data=old_data,

			




	
	
	
		
 
                         repo_groups=c.repo_groups_choices,

			




	
	
	
		
 
                         landing_revs=c.landing_revs_choices)()

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            form_result = _form.to_python(dict(request.POST))

			




	
	
	
		
 
            repo = repo_model.update(repo_name, **form_result)

			




	
	
	
		
 
            ScmModel().mark_for_invalidation(repo_name)

			




	
	
	
		
 
            h.flash(_('Repository %s updated successfully') % repo_name,

			




	
	
	
		
 
                    category='success')

			




	
	
	
		
 
            changed_name = repo.repo_name

			




	
	
	
		
 
            action_logger(self.authuser, 'admin_updated_repo',

			




	
	
	
		
 
                              changed_name, self.ip_addr, self.sa)

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 
        except formencode.Invalid, errors:

			




	
	
	
		
 
            log.info(errors)

			




	
	
	
		
 
            defaults = self.__load_data(repo_name)

			




	
	
	
		
 
            defaults.update(errors.value)

			




        

    


    
    

    

        

                

                    kallithea/tests/functional/test_admin_repos.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -590,49 +590,49 @@ class _BaseTest(object):

			




	
	
	
		
 
        user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')

			




	
	
	
		
 
        user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')

			




	
	
	
		
 
        user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')

			




	
	
	
		
 
        user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')

			




	
	
	
		
 

			




	
	
	
		
 
        # disable on regular user

			




	
	
	
		
 
        user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')

			




	
	
	
		
 
        user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')

			




	
	
	
		
 
        user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')

			




	
	
	
		
 
        user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
        user = User.get(usr['user_id'])

			




	
	
	
		
 

			




	
	
	
		
 
        repo_name = self.NEW_REPO+'no_perms'

			




	
	
	
		
 
        description = 'description for newly created repo'

			




	
	
	
		
 
        response = self.app.post(url('repos'),

			




	
	
	
		
 
                        fixture._get_repo_create_params(repo_private=False,

			




	
	
	
		
 
                                                repo_name=repo_name,

			




	
	
	
		
 
                                                repo_type=self.REPO_TYPE,

			




	
	
	
		
 
                                                repo_description=description,

			




	
	
	
		
 
                                                _authentication_token=self.authentication_token()))

			




	
	
	
		
 

			




	
	
	
		
 
        response.mustcontain('no permission to create repository in root location')

			




	
	
	
		
 
        response.mustcontain('<span class="error-message">Invalid value</span>')

			




	
	
	
		
 

			




	
	
	
		
 
        RepoModel().delete(repo_name)

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
    @mock.patch.object(RepoModel, '_create_filesystem_repo', error_function)

			




	
	
	
		
 
    def test_create_repo_when_filesystem_op_fails(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        repo_name = self.NEW_REPO

			




	
	
	
		
 
        description = 'description for newly created repo'

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.post(url('repos'),

			




	
	
	
		
 
                        fixture._get_repo_create_params(repo_private=False,

			




	
	
	
		
 
                                                repo_name=repo_name,

			




	
	
	
		
 
                                                repo_type=self.REPO_TYPE,

			




	
	
	
		
 
                                                repo_description=description,

			




	
	
	
		
 
                                                _authentication_token=self.authentication_token()))

			




	
	
	
		
 

			




	
	
	
		
 
        self.checkSessionFlash(response,

			




	
	
	
		
 
                               'Error creating repository %s' % repo_name)

			




	
	
	
		
 
        # repo must not be in db

			




	
	
	
		
 
        repo = Repository.get_by_repo_name(repo_name)

			




	
	
	
		
 
        self.assertEqual(repo, None)

			




	
	
	
		
 

			




	
	
	
		
 
        # repo must not be in filesystem !

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








    
        
    
    
        This site is powered by
            Kallithea 0.7.0,
        which is
        © 2010–2025 by various authors & licensed under GPLv3.