Changeset - 1522c37cbc62
[Not reviewed]
default
0 4 0
Mads Kiilerich - 9 years ago 2016-08-04 14:23:36
madski@unity3d.com
routing: use POST to 'update_users_group' instead of PUT
4 files changed with 5 insertions and 6 deletions:
0 comments (0 inline, 0 general)
kallithea/config/routing.py
Show inline comments
 
@@ -35,385 +35,385 @@ def make_map(config):
 
    from kallithea.lib.utils import (is_valid_repo, is_valid_repo_group,
 
                                     get_repo_by_id)
 

	
 
    def check_repo(environ, match_dict):
 
        """
 
        check for valid repository for proper 404 handling
 

	
 
        :param environ:
 
        :param match_dict:
 
        """
 
        repo_name = match_dict.get('repo_name')
 

	
 
        if match_dict.get('f_path'):
 
            #fix for multiple initial slashes that causes errors
 
            match_dict['f_path'] = match_dict['f_path'].lstrip('/')
 

	
 
        by_id_match = get_repo_by_id(repo_name)
 
        if by_id_match:
 
            repo_name = by_id_match
 
            match_dict['repo_name'] = repo_name
 

	
 
        return is_valid_repo(repo_name, config['base_path'])
 

	
 
    def check_group(environ, match_dict):
 
        """
 
        check for valid repository group for proper 404 handling
 

	
 
        :param environ:
 
        :param match_dict:
 
        """
 
        repo_group_name = match_dict.get('group_name')
 
        return is_valid_repo_group(repo_group_name, config['base_path'])
 

	
 
    def check_group_skip_path(environ, match_dict):
 
        """
 
        check for valid repository group for proper 404 handling, but skips
 
        verification of existing path
 

	
 
        :param environ:
 
        :param match_dict:
 
        """
 
        repo_group_name = match_dict.get('group_name')
 
        return is_valid_repo_group(repo_group_name, config['base_path'],
 
                                   skip_path_check=True)
 

	
 
    def check_user_group(environ, match_dict):
 
        """
 
        check for valid user group for proper 404 handling
 

	
 
        :param environ:
 
        :param match_dict:
 
        """
 
        return True
 

	
 
    def check_int(environ, match_dict):
 
        return match_dict.get('id').isdigit()
 

	
 
    # The ErrorController route (handles 404/500 error pages); it should
 
    # likely stay at the top, ensuring it can always be resolved
 
    rmap.connect('/error/{action}', controller='error')
 
    rmap.connect('/error/{action}/{id}', controller='error')
 

	
 
    #==========================================================================
 
    # CUSTOM ROUTES HERE
 
    #==========================================================================
 

	
 
    #MAIN PAGE
 
    rmap.connect('home', '/', controller='home', action='index')
 
    rmap.connect('about', '/about', controller='home', action='about')
 
    rmap.connect('repo_switcher_data', '/_repos', controller='home',
 
                 action='repo_switcher_data')
 

	
 
    rmap.connect('rst_help',
 
                 "http://docutils.sourceforge.net/docs/user/rst/quickref.html",
 
                 _static=True)
 
    rmap.connect('kallithea_project_url', "https://kallithea-scm.org/", _static=True)
 
    rmap.connect('issues_url', 'https://bitbucket.org/conservancy/kallithea/issues', _static=True)
 

	
 
    #ADMIN REPOSITORY ROUTES
 
    with rmap.submapper(path_prefix=ADMIN_PREFIX,
 
                        controller='admin/repos') as m:
 
        m.connect("repos", "/repos",
 
                  action="create", conditions=dict(method=["POST"]))
 
        m.connect("repos", "/repos",
 
                  action="index", conditions=dict(method=["GET"]))
 
        m.connect("new_repo", "/create_repository",
 
                  action="create_repository", conditions=dict(method=["GET"]))
 
        m.connect("update_repo", "/repos/{repo_name:.*?}",
 
                  action="update", conditions=dict(method=["POST"],
 
                  function=check_repo))
 
        m.connect("delete_repo", "/repos/{repo_name:.*?}/delete",
 
                  action="delete", conditions=dict(method=["POST"]))
 

	
 
    #ADMIN REPOSITORY GROUPS ROUTES
 
    with rmap.submapper(path_prefix=ADMIN_PREFIX,
 
                        controller='admin/repo_groups') as m:
 
        m.connect("repos_groups", "/repo_groups",
 
                  action="create", conditions=dict(method=["POST"]))
 
        m.connect("repos_groups", "/repo_groups",
 
                  action="index", conditions=dict(method=["GET"]))
 
        m.connect("new_repos_group", "/repo_groups/new",
 
                  action="new", conditions=dict(method=["GET"]))
 
        m.connect("update_repos_group", "/repo_groups/{group_name:.*?}",
 
                  action="update", conditions=dict(method=["POST"],
 
                                                   function=check_group))
 

	
 
        m.connect("repos_group", "/repo_groups/{group_name:.*?}",
 
                  action="show", conditions=dict(method=["GET"],
 
                                                 function=check_group))
 

	
 
        #EXTRAS REPO GROUP ROUTES
 
        m.connect("edit_repo_group", "/repo_groups/{group_name:.*?}/edit",
 
                  action="edit",
 
                  conditions=dict(method=["GET"], function=check_group))
 

	
 
        m.connect("edit_repo_group_advanced", "/repo_groups/{group_name:.*?}/edit/advanced",
 
                  action="edit_repo_group_advanced",
 
                  conditions=dict(method=["GET"], function=check_group))
 

	
 
        m.connect("edit_repo_group_perms", "/repo_groups/{group_name:.*?}/edit/permissions",
 
                  action="edit_repo_group_perms",
 
                  conditions=dict(method=["GET"], function=check_group))
 
        m.connect("edit_repo_group_perms_update", "/repo_groups/{group_name:.*?}/edit/permissions",
 
                  action="update_perms",
 
                  conditions=dict(method=["POST"], function=check_group))
 
        m.connect("edit_repo_group_perms_delete", "/repo_groups/{group_name:.*?}/edit/permissions/delete",
 
                  action="delete_perms",
 
                  conditions=dict(method=["POST"], function=check_group))
 

	
 
        m.connect("delete_repo_group", "/repo_groups/{group_name:.*?}/delete",
 
                  action="delete", conditions=dict(method=["POST"],
 
                                                   function=check_group_skip_path))
 

	
 

	
 
    #ADMIN USER ROUTES
 
    with rmap.submapper(path_prefix=ADMIN_PREFIX,
 
                        controller='admin/users') as m:
 
        m.connect("users", "/users",
 
                  action="create", conditions=dict(method=["POST"]))
 
        m.connect("users", "/users",
 
                  action="index", conditions=dict(method=["GET"]))
 
        m.connect("formatted_users", "/users.{format}",
 
                  action="index", conditions=dict(method=["GET"]))
 
        m.connect("new_user", "/users/new",
 
                  action="new", conditions=dict(method=["GET"]))
 
        m.connect("update_user", "/users/{id}",
 
                  action="update", conditions=dict(method=["POST"]))
 
        m.connect("delete_user", "/users/{id}/delete",
 
                  action="delete", conditions=dict(method=["POST"]))
 
        m.connect("edit_user", "/users/{id}/edit",
 
                  action="edit", conditions=dict(method=["GET"]))
 

	
 
        #EXTRAS USER ROUTES
 
        m.connect("edit_user_advanced", "/users/{id}/edit/advanced",
 
                  action="edit_advanced", conditions=dict(method=["GET"]))
 

	
 
        m.connect("edit_user_api_keys", "/users/{id}/edit/api_keys",
 
                  action="edit_api_keys", conditions=dict(method=["GET"]))
 
        m.connect("edit_user_api_keys", "/users/{id}/edit/api_keys",
 
                  action="add_api_key", conditions=dict(method=["POST"]))
 
        m.connect("edit_user_api_keys_delete", "/users/{id}/edit/api_keys/delete",
 
                  action="delete_api_key", conditions=dict(method=["POST"]))
 

	
 
        m.connect("edit_user_perms", "/users/{id}/edit/permissions",
 
                  action="edit_perms", conditions=dict(method=["GET"]))
 
        m.connect("edit_user_perms", "/users/{id}/edit/permissions",
 
                  action="update_perms", conditions=dict(method=["PUT"]))
 

	
 
        m.connect("edit_user_emails", "/users/{id}/edit/emails",
 
                  action="edit_emails", conditions=dict(method=["GET"]))
 
        m.connect("edit_user_emails_update", "/users/{id}/edit/emails",
 
                  action="add_email", conditions=dict(method=["POST"]))
 
        m.connect("edit_user_emails_delete", "/users/{id}/edit/emails/delete",
 
                  action="delete_email", conditions=dict(method=["POST"]))
 

	
 
        m.connect("edit_user_ips", "/users/{id}/edit/ips",
 
                  action="edit_ips", conditions=dict(method=["GET"]))
 
        m.connect("edit_user_ips_update", "/users/{id}/edit/ips",
 
                  action="add_ip", conditions=dict(method=["POST"]))
 
        m.connect("edit_user_ips_delete", "/users/{id}/edit/ips/delete",
 
                  action="delete_ip", conditions=dict(method=["POST"]))
 

	
 
    #ADMIN USER GROUPS REST ROUTES
 
    with rmap.submapper(path_prefix=ADMIN_PREFIX,
 
                        controller='admin/user_groups') as m:
 
        m.connect("users_groups", "/user_groups",
 
                  action="create", conditions=dict(method=["POST"]))
 
        m.connect("users_groups", "/user_groups",
 
                  action="index", conditions=dict(method=["GET"]))
 
        m.connect("new_users_group", "/user_groups/new",
 
                  action="new", conditions=dict(method=["GET"]))
 
        m.connect("update_users_group", "/user_groups/{id}",
 
                  action="update", conditions=dict(method=["PUT"]))
 
                  action="update", conditions=dict(method=["POST"]))
 
        m.connect("delete_users_group", "/user_groups/{id}/delete",
 
                  action="delete", conditions=dict(method=["POST"]))
 
        m.connect("edit_users_group", "/user_groups/{id}/edit",
 
                  action="edit", conditions=dict(method=["GET"]),
 
                  function=check_user_group)
 

	
 
        #EXTRAS USER GROUP ROUTES
 
        m.connect("edit_user_group_default_perms", "/user_groups/{id}/edit/default_perms",
 
                  action="edit_default_perms", conditions=dict(method=["GET"]))
 
        m.connect("edit_user_group_default_perms", "/user_groups/{id}/edit/default_perms",
 
                  action="update_default_perms", conditions=dict(method=["PUT"]))
 

	
 

	
 
        m.connect("edit_user_group_perms", "/user_groups/{id}/edit/perms",
 
                  action="edit_perms", conditions=dict(method=["GET"]))
 
        m.connect("edit_user_group_perms", "/user_groups/{id}/edit/perms",
 
                  action="update_perms", conditions=dict(method=["PUT"]))
 
        m.connect("edit_user_group_perms_delete", "/user_groups/{id}/edit/perms/delete",
 
                  action="delete_perms", conditions=dict(method=["POST"]))
 

	
 
        m.connect("edit_user_group_advanced", "/user_groups/{id}/edit/advanced",
 
                  action="edit_advanced", conditions=dict(method=["GET"]))
 

	
 
        m.connect("edit_user_group_members", "/user_groups/{id}/edit/members",
 
                  action="edit_members", conditions=dict(method=["GET"]))
 

	
 

	
 

	
 
    #ADMIN PERMISSIONS ROUTES
 
    with rmap.submapper(path_prefix=ADMIN_PREFIX,
 
                        controller='admin/permissions') as m:
 
        m.connect("admin_permissions", "/permissions",
 
                  action="permission_globals", conditions=dict(method=["POST"]))
 
        m.connect("admin_permissions", "/permissions",
 
                  action="permission_globals", conditions=dict(method=["GET"]))
 

	
 
        m.connect("admin_permissions_ips", "/permissions/ips",
 
                  action="permission_ips", conditions=dict(method=["GET"]))
 

	
 
        m.connect("admin_permissions_perms", "/permissions/perms",
 
                  action="permission_perms", conditions=dict(method=["GET"]))
 

	
 

	
 
    #ADMIN DEFAULTS ROUTES
 
    with rmap.submapper(path_prefix=ADMIN_PREFIX,
 
                        controller='admin/defaults') as m:
 
        m.connect('defaults', 'defaults',
 
                  action="index")
 
        m.connect('defaults_update', 'defaults/{id}/update',
 
                  action="update", conditions=dict(method=["POST"]))
 

	
 
    #ADMIN AUTH SETTINGS
 
    rmap.connect('auth_settings', '%s/auth' % ADMIN_PREFIX,
 
                 controller='admin/auth_settings', action='auth_settings',
 
                 conditions=dict(method=["POST"]))
 
    rmap.connect('auth_home', '%s/auth' % ADMIN_PREFIX,
 
                 controller='admin/auth_settings')
 

	
 
    #ADMIN SETTINGS ROUTES
 
    with rmap.submapper(path_prefix=ADMIN_PREFIX,
 
                        controller='admin/settings') as m:
 
        m.connect("admin_settings", "/settings",
 
                  action="settings_vcs", conditions=dict(method=["POST"]))
 
        m.connect("admin_settings", "/settings",
 
                  action="settings_vcs", conditions=dict(method=["GET"]))
 

	
 
        m.connect("admin_settings_mapping", "/settings/mapping",
 
                  action="settings_mapping", conditions=dict(method=["POST"]))
 
        m.connect("admin_settings_mapping", "/settings/mapping",
 
                  action="settings_mapping", conditions=dict(method=["GET"]))
 

	
 
        m.connect("admin_settings_global", "/settings/global",
 
                  action="settings_global", conditions=dict(method=["POST"]))
 
        m.connect("admin_settings_global", "/settings/global",
 
                  action="settings_global", conditions=dict(method=["GET"]))
 

	
 
        m.connect("admin_settings_visual", "/settings/visual",
 
                  action="settings_visual", conditions=dict(method=["POST"]))
 
        m.connect("admin_settings_visual", "/settings/visual",
 
                  action="settings_visual", conditions=dict(method=["GET"]))
 

	
 
        m.connect("admin_settings_email", "/settings/email",
 
                  action="settings_email", conditions=dict(method=["POST"]))
 
        m.connect("admin_settings_email", "/settings/email",
 
                  action="settings_email", conditions=dict(method=["GET"]))
 

	
 
        m.connect("admin_settings_hooks", "/settings/hooks",
 
                  action="settings_hooks", conditions=dict(method=["POST"]))
 
        m.connect("admin_settings_hooks_delete", "/settings/hooks/delete",
 
                  action="settings_hooks", conditions=dict(method=["POST"]))
 
        m.connect("admin_settings_hooks", "/settings/hooks",
 
                  action="settings_hooks", conditions=dict(method=["GET"]))
 

	
 
        m.connect("admin_settings_search", "/settings/search",
 
                  action="settings_search", conditions=dict(method=["POST"]))
 
        m.connect("admin_settings_search", "/settings/search",
 
                  action="settings_search", conditions=dict(method=["GET"]))
 

	
 
        m.connect("admin_settings_system", "/settings/system",
 
                  action="settings_system", conditions=dict(method=["POST"]))
 
        m.connect("admin_settings_system", "/settings/system",
 
                  action="settings_system", conditions=dict(method=["GET"]))
 
        m.connect("admin_settings_system_update", "/settings/system/updates",
 
                  action="settings_system_update", conditions=dict(method=["GET"]))
 

	
 
    #ADMIN MY ACCOUNT
 
    with rmap.submapper(path_prefix=ADMIN_PREFIX,
 
                        controller='admin/my_account') as m:
 

	
 
        m.connect("my_account", "/my_account",
 
                  action="my_account", conditions=dict(method=["GET"]))
 
        m.connect("my_account", "/my_account",
 
                  action="my_account", conditions=dict(method=["POST"]))
 

	
 
        m.connect("my_account_password", "/my_account/password",
 
                  action="my_account_password", conditions=dict(method=["GET"]))
 
        m.connect("my_account_password", "/my_account/password",
 
                  action="my_account_password", conditions=dict(method=["POST"]))
 

	
 
        m.connect("my_account_repos", "/my_account/repos",
 
                  action="my_account_repos", conditions=dict(method=["GET"]))
 

	
 
        m.connect("my_account_watched", "/my_account/watched",
 
                  action="my_account_watched", conditions=dict(method=["GET"]))
 

	
 
        m.connect("my_account_perms", "/my_account/perms",
 
                  action="my_account_perms", conditions=dict(method=["GET"]))
 

	
 
        m.connect("my_account_emails", "/my_account/emails",
 
                  action="my_account_emails", conditions=dict(method=["GET"]))
 
        m.connect("my_account_emails", "/my_account/emails",
 
                  action="my_account_emails_add", conditions=dict(method=["POST"]))
 
        m.connect("my_account_emails_delete", "/my_account/emails/delete",
 
                  action="my_account_emails_delete", conditions=dict(method=["POST"]))
 

	
 
        m.connect("my_account_api_keys", "/my_account/api_keys",
 
                  action="my_account_api_keys", conditions=dict(method=["GET"]))
 
        m.connect("my_account_api_keys", "/my_account/api_keys",
 
                  action="my_account_api_keys_add", conditions=dict(method=["POST"]))
 
        m.connect("my_account_api_keys_delete", "/my_account/api_keys/delete",
 
                  action="my_account_api_keys_delete", conditions=dict(method=["POST"]))
 

	
 
    #NOTIFICATION REST ROUTES
 
    with rmap.submapper(path_prefix=ADMIN_PREFIX,
 
                        controller='admin/notifications') as m:
 
        m.connect("notifications", "/notifications",
 
                  action="index", conditions=dict(method=["GET"]))
 
        m.connect("notifications_mark_all_read", "/notifications/mark_all_read",
 
                  action="mark_all_read", conditions=dict(method=["GET"]))
 
        m.connect("formatted_notifications", "/notifications.{format}",
 
                  action="index", conditions=dict(method=["GET"]))
 
        m.connect("notification_update", "/notifications/{notification_id}/update",
 
                  action="update", conditions=dict(method=["POST"]))
 
        m.connect("notification_delete", "/notifications/{notification_id}/delete",
 
                  action="delete", conditions=dict(method=["POST"]))
 
        m.connect("notification", "/notifications/{notification_id}",
 
                  action="show", conditions=dict(method=["GET"]))
 
        m.connect("formatted_notification", "/notifications/{notification_id}.{format}",
 
                  action="show", conditions=dict(method=["GET"]))
 

	
 
    #ADMIN GIST
 
    with rmap.submapper(path_prefix=ADMIN_PREFIX,
 
                        controller='admin/gists') as m:
 
        m.connect("gists", "/gists",
 
                  action="create", conditions=dict(method=["POST"]))
 
        m.connect("gists", "/gists",
 
                  action="index", conditions=dict(method=["GET"]))
 
        m.connect("new_gist", "/gists/new",
 
                  action="new", conditions=dict(method=["GET"]))
 

	
 

	
 
        m.connect("gist_delete", "/gists/{gist_id}/delete",
 
                  action="delete", conditions=dict(method=["POST"]))
 
        m.connect("edit_gist", "/gists/{gist_id}/edit",
 
                  action="edit", conditions=dict(method=["GET", "POST"]))
 
        m.connect("edit_gist_check_revision", "/gists/{gist_id}/edit/check_revision",
 
                  action="check_revision", conditions=dict(method=["POST"]))
 

	
 

	
 
        m.connect("gist", "/gists/{gist_id}",
 
                  action="show", conditions=dict(method=["GET"]))
 
        m.connect("gist_rev", "/gists/{gist_id}/{revision}",
 
                  revision="tip",
 
                  action="show", conditions=dict(method=["GET"]))
 
        m.connect("formatted_gist", "/gists/{gist_id}/{revision}/{format}",
 
                  revision="tip",
 
                  action="show", conditions=dict(method=["GET"]))
 
        m.connect("formatted_gist_file", "/gists/{gist_id}/{revision}/{format}/{f_path:.*}",
 
                  revision='tip',
 
                  action="show", conditions=dict(method=["GET"]))
 

	
 
    #ADMIN MAIN PAGES
kallithea/controllers/admin/user_groups.py
Show inline comments
 
@@ -5,385 +5,384 @@
 
# (at your option) any later version.
 
#
 
# 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 General Public License
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
"""
 
kallithea.controllers.admin.user_groups
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

	
 
User Groups crud controller for pylons
 

	
 
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: Jan 25, 2011
 
: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 pylons import request, tmpl_context as c, url, config
 
from pylons.i18n.translation import _
 
from webob.exc import HTTPFound
 

	
 
from sqlalchemy.orm import joinedload
 
from sqlalchemy.sql.expression import func
 
from webob.exc import HTTPInternalServerError
 

	
 
import kallithea
 
from kallithea.lib import helpers as h
 
from kallithea.lib.exceptions import UserGroupsAssignedException, \
 
    RepoGroupAssignmentError
 
from kallithea.lib.utils2 import safe_unicode, safe_int
 
from kallithea.lib.auth import LoginRequired, \
 
    HasUserGroupPermissionAnyDecorator, HasPermissionAnyDecorator
 
from kallithea.lib.base import BaseController, render
 
from kallithea.model.scm import UserGroupList
 
from kallithea.model.user_group import UserGroupModel
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.db import User, UserGroup, UserGroupToPerm, \
 
    UserGroupRepoToPerm, UserGroupRepoGroupToPerm
 
from kallithea.model.forms import UserGroupForm, UserGroupPermsForm, \
 
    CustomDefaultPermissionsForm
 
from kallithea.model.meta import Session
 
from kallithea.lib.utils import action_logger
 
from kallithea.lib.compat import json
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
class UserGroupsController(BaseController):
 
    """REST Controller styled on the Atom Publishing Protocol"""
 

	
 
    @LoginRequired()
 
    def __before__(self):
 
        super(UserGroupsController, self).__before__()
 
        c.available_permissions = config['available_permissions']
 

	
 
    def __load_data(self, user_group_id):
 
        c.group_members_obj = sorted((x.user for x in c.user_group.members),
 
                                     key=lambda u: u.username.lower())
 

	
 
        c.group_members = [(x.user_id, x.username) for x in c.group_members_obj]
 
        c.available_members = sorted(((x.user_id, x.username) for x in
 
                                      User.query().all()),
 
                                     key=lambda u: u[1].lower())
 

	
 
    def __load_defaults(self, user_group_id):
 
        """
 
        Load defaults settings for edit, and update
 

	
 
        :param user_group_id:
 
        """
 
        user_group = UserGroup.get_or_404(user_group_id)
 
        data = user_group.get_dict()
 
        return data
 

	
 
    def index(self, format='html'):
 
        _list = UserGroup.query() \
 
                        .order_by(func.lower(UserGroup.users_group_name)) \
 
                        .all()
 
        group_iter = UserGroupList(_list, perm_set=['usergroup.admin'])
 
        user_groups_data = []
 
        total_records = len(group_iter)
 
        _tmpl_lookup = kallithea.CONFIG['pylons.app_globals'].mako_lookup
 
        template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
 

	
 
        user_group_name = lambda user_group_id, user_group_name: (
 
            template.get_def("user_group_name")
 
            .render(user_group_id, user_group_name, _=_, h=h, c=c)
 
        )
 
        user_group_actions = lambda user_group_id, user_group_name: (
 
            template.get_def("user_group_actions")
 
            .render(user_group_id, user_group_name, _=_, h=h, c=c)
 
        )
 
        for user_gr in group_iter:
 

	
 
            user_groups_data.append({
 
                "raw_name": user_gr.users_group_name,
 
                "group_name": user_group_name(user_gr.users_group_id,
 
                                              user_gr.users_group_name),
 
                "desc": h.escape(user_gr.user_group_description),
 
                "members": len(user_gr.members),
 
                "active": h.boolicon(user_gr.users_group_active),
 
                "owner": h.person(user_gr.user.username),
 
                "action": user_group_actions(user_gr.users_group_id, user_gr.users_group_name)
 
            })
 

	
 
        c.data = json.dumps({
 
            "totalRecords": total_records,
 
            "startIndex": 0,
 
            "sort": None,
 
            "dir": "asc",
 
            "records": user_groups_data
 
        })
 

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

	
 
    @HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')
 
    def create(self):
 
        users_group_form = UserGroupForm()()
 
        try:
 
            form_result = users_group_form.to_python(dict(request.POST))
 
            ug = UserGroupModel().create(name=form_result['users_group_name'],
 
                                         description=form_result['user_group_description'],
 
                                         owner=self.authuser.user_id,
 
                                         active=form_result['users_group_active'])
 

	
 
            gr = form_result['users_group_name']
 
            action_logger(self.authuser,
 
                          'admin_created_users_group:%s' % gr,
 
                          None, self.ip_addr, self.sa)
 
            h.flash(h.literal(_('Created user group %s') % h.link_to(h.escape(gr), url('edit_users_group', id=ug.users_group_id))),
 
                category='success')
 
            Session().commit()
 
        except formencode.Invalid as errors:
 
            return htmlfill.render(
 
                render('admin/user_groups/user_group_add.html'),
 
                defaults=errors.value,
 
                errors=errors.error_dict or {},
 
                prefix_error=False,
 
                encoding="UTF-8",
 
                force_defaults=False)
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            h.flash(_('Error occurred during creation of user group %s') \
 
                    % request.POST.get('users_group_name'), category='error')
 

	
 
        raise HTTPFound(location=url('users_groups'))
 

	
 
    @HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')
 
    def new(self, format='html'):
 
        return render('admin/user_groups/user_group_add.html')
 

	
 
    @HasUserGroupPermissionAnyDecorator('usergroup.admin')
 
    def update(self, id):
 
        c.user_group = UserGroup.get_or_404(id)
 
        c.active = 'settings'
 
        self.__load_data(id)
 

	
 
        available_members = [safe_unicode(x[0]) for x in c.available_members]
 

	
 
        users_group_form = UserGroupForm(edit=True,
 
                                         old_data=c.user_group.get_dict(),
 
                                         available_members=available_members)()
 

	
 
        try:
 
            form_result = users_group_form.to_python(request.POST)
 
            UserGroupModel().update(c.user_group, form_result)
 
            gr = form_result['users_group_name']
 
            action_logger(self.authuser,
 
                          'admin_updated_users_group:%s' % gr,
 
                          None, self.ip_addr, self.sa)
 
            h.flash(_('Updated user group %s') % gr, category='success')
 
            Session().commit()
 
        except formencode.Invalid as errors:
 
            ug_model = UserGroupModel()
 
            defaults = errors.value
 
            e = errors.error_dict or {}
 
            defaults.update({
 
                'create_repo_perm': ug_model.has_perm(id,
 
                                                      'hg.create.repository'),
 
                'fork_repo_perm': ug_model.has_perm(id,
 
                                                    'hg.fork.repository'),
 
                '_method': 'put'
 
            })
 

	
 
            return htmlfill.render(
 
                render('admin/user_groups/user_group_edit.html'),
 
                defaults=defaults,
 
                errors=e,
 
                prefix_error=False,
 
                encoding="UTF-8",
 
                force_defaults=False)
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            h.flash(_('Error occurred during update of user group %s') \
 
                    % request.POST.get('users_group_name'), category='error')
 

	
 
        raise HTTPFound(location=url('edit_users_group', id=id))
 

	
 
    @HasUserGroupPermissionAnyDecorator('usergroup.admin')
 
    def delete(self, id):
 
        usr_gr = UserGroup.get_or_404(id)
 
        try:
 
            UserGroupModel().delete(usr_gr)
 
            Session().commit()
 
            h.flash(_('Successfully deleted user group'), category='success')
 
        except UserGroupsAssignedException as e:
 
            h.flash(e, category='error')
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            h.flash(_('An error occurred during deletion of user group'),
 
                    category='error')
 
        raise HTTPFound(location=url('users_groups'))
 

	
 
    @HasUserGroupPermissionAnyDecorator('usergroup.admin')
 
    def edit(self, id, format='html'):
 
        c.user_group = UserGroup.get_or_404(id)
 
        c.active = 'settings'
 
        self.__load_data(id)
 

	
 
        defaults = self.__load_defaults(id)
 

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

	
 
    @HasUserGroupPermissionAnyDecorator('usergroup.admin')
 
    def edit_perms(self, id):
 
        c.user_group = UserGroup.get_or_404(id)
 
        c.active = 'perms'
 

	
 
        repo_model = RepoModel()
 
        c.users_array = repo_model.get_users_js()
 
        c.user_groups_array = repo_model.get_user_groups_js()
 

	
 
        defaults = {}
 
        # fill user group users
 
        for p in c.user_group.user_user_group_to_perm:
 
            defaults.update({'u_perm_%s' % p.user.username:
 
                             p.permission.permission_name})
 

	
 
        for p in c.user_group.user_group_user_group_to_perm:
 
            defaults.update({'g_perm_%s' % p.user_group.users_group_name:
 
                             p.permission.permission_name})
 

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

	
 
    @HasUserGroupPermissionAnyDecorator('usergroup.admin')
 
    def update_perms(self, id):
 
        """
 
        grant permission for given usergroup
 

	
 
        :param id:
 
        """
 
        user_group = UserGroup.get_or_404(id)
 
        form = UserGroupPermsForm()().to_python(request.POST)
 

	
 
        # set the permissions !
 
        try:
 
            UserGroupModel()._update_permissions(user_group, form['perms_new'],
 
                                                 form['perms_updates'])
 
        except RepoGroupAssignmentError:
 
            h.flash(_('Target group cannot be the same'), category='error')
 
            raise HTTPFound(location=url('edit_user_group_perms', id=id))
 
        #TODO: implement this
 
        #action_logger(self.authuser, 'admin_changed_repo_permissions',
 
        #              repo_name, self.ip_addr, self.sa)
 
        Session().commit()
 
        h.flash(_('User group permissions updated'), category='success')
 
        raise HTTPFound(location=url('edit_user_group_perms', id=id))
 

	
 
    @HasUserGroupPermissionAnyDecorator('usergroup.admin')
 
    def delete_perms(self, id):
 
        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 not c.authuser.is_admin:
 
                if obj_type == 'user' and c.authuser.user_id == obj_id:
 
                    msg = _('Cannot revoke permission for yourself as admin')
 
                    h.flash(msg, category='warning')
 
                    raise Exception('revoke admin permission on self')
 
            if obj_type == 'user':
 
                UserGroupModel().revoke_user_permission(user_group=id,
 
                                                        user=obj_id)
 
            elif obj_type == 'user_group':
 
                UserGroupModel().revoke_user_group_permission(target_user_group=id,
 
                                                              user_group=obj_id)
 
            Session().commit()
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            h.flash(_('An error occurred during revoking of permission'),
 
                    category='error')
 
            raise HTTPInternalServerError()
 

	
 
    @HasUserGroupPermissionAnyDecorator('usergroup.admin')
 
    def edit_default_perms(self, id):
 
        c.user_group = UserGroup.get_or_404(id)
 
        c.active = 'default_perms'
 

	
 
        permissions = {
 
            'repositories': {},
 
            'repositories_groups': {}
 
        }
 
        ugroup_repo_perms = UserGroupRepoToPerm.query() \
 
            .options(joinedload(UserGroupRepoToPerm.permission)) \
 
            .options(joinedload(UserGroupRepoToPerm.repository)) \
 
            .filter(UserGroupRepoToPerm.users_group_id == id) \
 
            .all()
 

	
 
        for gr in ugroup_repo_perms:
 
            permissions['repositories'][gr.repository.repo_name]  \
 
                = gr.permission.permission_name
 

	
 
        ugroup_group_perms = UserGroupRepoGroupToPerm.query() \
 
            .options(joinedload(UserGroupRepoGroupToPerm.permission)) \
 
            .options(joinedload(UserGroupRepoGroupToPerm.group)) \
 
            .filter(UserGroupRepoGroupToPerm.users_group_id == id) \
 
            .all()
 

	
 
        for gr in ugroup_group_perms:
 
            permissions['repositories_groups'][gr.group.group_name] \
 
                = gr.permission.permission_name
 
        c.permissions = permissions
 

	
 
        ug_model = UserGroupModel()
 

	
 
        defaults = c.user_group.get_dict()
 
        defaults.update({
 
            'create_repo_perm': ug_model.has_perm(c.user_group,
 
                                                  'hg.create.repository'),
 
            'create_user_group_perm': ug_model.has_perm(c.user_group,
 
                                                        'hg.usergroup.create.true'),
 
            'fork_repo_perm': ug_model.has_perm(c.user_group,
 
                                                'hg.fork.repository'),
 
        })
 

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

	
 
    @HasUserGroupPermissionAnyDecorator('usergroup.admin')
 
    def update_default_perms(self, id):
 
        user_group = UserGroup.get_or_404(id)
 

	
 
        try:
 
            form = CustomDefaultPermissionsForm()()
 
            form_result = form.to_python(request.POST)
 

	
 
            inherit_perms = form_result['inherit_default_permissions']
 
            user_group.inherit_default_permissions = inherit_perms
 
            Session().add(user_group)
 
            usergroup_model = UserGroupModel()
 

	
 
            defs = UserGroupToPerm.query() \
 
                .filter(UserGroupToPerm.users_group == user_group) \
 
                .all()
 
            for ug in defs:
 
                Session().delete(ug)
 

	
kallithea/templates/admin/user_groups/user_group_edit_settings.html
Show inline comments
 
${h.form(url('users_group', id=c.user_group.users_group_id),method='put', id='edit_users_group')}
 
${h.form(url('update_users_group', id=c.user_group.users_group_id), id='edit_users_group')}
 
    <div class="form">
 
        <!-- fields -->
 
            <div class="fields">
 
                 <div class="field">
 
                    <div class="label">
 
                        <label for="users_group_name">${_('Group name')}:</label>
 
                    </div>
 
                    <div class="input">
 
                        ${h.text('users_group_name',class_='large')}
 
                    </div>
 
                 </div>
 
                 <div class="field">
 
                    <div class="label label-textarea">
 
                        <label for="user_group_description">${_('Description')}:</label>
 
                    </div>
 
                    <div class="textarea-small editor">
 
                        ${h.textarea('user_group_description')}
 
                        <span class="help-block">${_('Short, optional description for this user group.')}</span>
 
                    </div>
 
                 </div>
 
                 <div class="field">
 
                    <div class="label label-checkbox">
 
                        <label for="users_group_active">${_('Active')}:</label>
 
                    </div>
 
                    <div class="checkboxes">
 
                        ${h.checkbox('users_group_active',value=True)}
 
                    </div>
 
                 </div>
 
                <div class="field">
 
                    <div class="label">
 
                        <label for="users_group_active">${_('Members')}:</label>
 
                    </div>
 
                    <div class="select">
 
                        <table>
 
                                <tr>
 
                                    <td>
 
                                        <div>
 
                                            <div style="float:left">
 
                                                <div class="text" style="padding: 0px 0px 6px;">${_('Chosen group members')}</div>
 
                                                ${h.select('users_group_members',[],c.group_members,multiple=True,size=8,style="min-width:210px")}
 
                                            </div>
 
                                            <div style="float:left;width:20px;padding-top:50px">
 
                                                <i style="cursor:pointer; font-size: 16px" id="add_element" class="icon-left-open" title="Choose selected available"></i>
 
                                                <div style="height:50px"></div>
 
                                                <i style="cursor:pointer; font-size: 16px" id="remove_element" class="icon-right-open" title="Remove selected chosen"></i>
 
                                            </div>
 
                                            <div style="float:left">
 
                                                 <div class="text" style="padding: 0px 0px 6px;">${_('Available members')}</div>
 
                                                 ${h.select('available_members',[],c.available_members,multiple=True,size=8,style="min-width:210px")}
 
                                            </div>
 
                                        </div>
 
                                    </td>
 
                                </tr>
 
                        </table>
 
                    </div>
 

	
 
                </div>
 
                <div class="buttons">
 
                  ${h.submit('Save',_('Save'),class_="btn")}
 
                </div>
 
            </div>
 
    </div>
 
${h.end_form()}
 
<script type="text/javascript">
 
  MultiSelectWidget('users_group_members','available_members','edit_users_group');
 
</script>
kallithea/tests/functional/test_admin_user_groups.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
from kallithea.tests import *
 
from kallithea.model.db import UserGroup, UserGroupToPerm, Permission
 
from kallithea.model.meta import Session
 

	
 
TEST_USER_GROUP = u'admins_test'
 

	
 

	
 
class TestAdminUsersGroupsController(TestController):
 

	
 
    def test_index(self):
 
        self.log_user()
 
        response = self.app.get(url('users_groups'))
 
        # Test response...
 

	
 
    def test_create(self):
 
        self.log_user()
 
        users_group_name = TEST_USER_GROUP
 
        response = self.app.post(url('users_groups'),
 
                                 {'users_group_name': users_group_name,
 
                                  'user_group_description': u'DESC',
 
                                  'active': True,
 
                                  '_authentication_token': self.authentication_token()})
 
        response.follow()
 

	
 
        self.checkSessionFlash(response,
 
                               'Created user group <a href="/_admin/user_groups/')
 
        self.checkSessionFlash(response,
 
                               '/edit">%s</a>' % TEST_USER_GROUP)
 

	
 
    def test_new(self):
 
        response = self.app.get(url('new_users_group'))
 

	
 
    def test_update(self):
 
        response = self.app.put(url('users_group', id=1), status=403)
 
        response = self.app.post(url('update_users_group', id=1), status=403)
 

	
 
    def test_update_browser_fakeout(self):
 
        response = self.app.post(url('users_group', id=1),
 
                                 params=dict(_method='put', _authentication_token=self.authentication_token()))
 
        response = self.app.post(url('update_users_group', id=1),
 
                                 params=dict(_authentication_token=self.authentication_token()))
 

	
 
    def test_delete(self):
 
        self.log_user()
 
        users_group_name = TEST_USER_GROUP + 'another'
 
        response = self.app.post(url('users_groups'),
 
                                 {'users_group_name':users_group_name,
 
                                  'user_group_description': u'DESC',
 
                                  'active': True,
 
                                  '_authentication_token': self.authentication_token()})
 
        response.follow()
 

	
 
        self.checkSessionFlash(response,
 
                               'Created user group ')
 

	
 
        gr = Session().query(UserGroup) \
 
            .filter(UserGroup.users_group_name == users_group_name).one()
 

	
 
        response = self.app.post(url('delete_users_group', id=gr.users_group_id),
 
            params={'_authentication_token': self.authentication_token()})
 

	
 
        gr = Session().query(UserGroup) \
 
            .filter(UserGroup.users_group_name == users_group_name).scalar()
 

	
 
        assert gr == None
 

	
 
    def test_default_perms_enable_repository_read_on_group(self):
 
        self.log_user()
 
        users_group_name = TEST_USER_GROUP + 'another2'
 
        response = self.app.post(url('users_groups'),
 
                                 {'users_group_name': users_group_name,
 
                                  'user_group_description': u'DESC',
 
                                  'active': True,
 
                                  '_authentication_token': self.authentication_token()})
 
        response.follow()
 

	
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        self.checkSessionFlash(response,
 
                               'Created user group ')
 
        ## ENABLE REPO CREATE ON A GROUP
 
        response = self.app.put(url('edit_user_group_default_perms',
 
                                    id=ug.users_group_id),
 
                                 {'create_repo_perm': True,
 
                                  '_authentication_token': self.authentication_token()})
 
        response.follow()
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        p = Permission.get_by_key('hg.create.repository')
 
        p2 = Permission.get_by_key('hg.usergroup.create.false')
 
        p3 = Permission.get_by_key('hg.fork.none')
 
        # check if user has this perms, they should be here since
 
        # defaults are on
 
        perms = UserGroupToPerm.query() \
 
            .filter(UserGroupToPerm.users_group == ug).all()
 

	
 
        assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
 
                    [ug.users_group_id, p2.permission_id],
 
                    [ug.users_group_id, p3.permission_id]])
 

	
 
        ## DISABLE REPO CREATE ON A GROUP
 
        response = self.app.put(
 
            url('edit_user_group_default_perms', id=ug.users_group_id),
 
            params={'_authentication_token': self.authentication_token()})
 

	
 
        response.follow()
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        p = Permission.get_by_key('hg.create.none')
 
        p2 = Permission.get_by_key('hg.usergroup.create.false')
 
        p3 = Permission.get_by_key('hg.fork.none')
 

	
 
        # check if user has this perms, they should be here since
 
        # defaults are on
 
        perms = UserGroupToPerm.query() \
 
            .filter(UserGroupToPerm.users_group == ug).all()
 

	
 
        assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
 
                    [ug.users_group_id, p2.permission_id],
 
                    [ug.users_group_id, p3.permission_id]])
 

	
 
        # DELETE !
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        ugid = ug.users_group_id
 
        response = self.app.post(url('delete_users_group', id=ug.users_group_id),
 
            params={'_authentication_token': self.authentication_token()})
 
        response = response.follow()
 
        gr = Session().query(UserGroup) \
 
            .filter(UserGroup.users_group_name == users_group_name).scalar()
 

	
 
        assert gr == None
 
        p = Permission.get_by_key('hg.create.repository')
 
        perms = UserGroupToPerm.query() \
 
            .filter(UserGroupToPerm.users_group_id == ugid).all()
 
        perms = [[x.users_group_id,
 
                  x.permission_id, ] for x in perms]
 
        assert perms == []
 

	
 
    def test_default_perms_enable_repository_fork_on_group(self):
 
        self.log_user()
 
        users_group_name = TEST_USER_GROUP + 'another2'
 
        response = self.app.post(url('users_groups'),
 
                                 {'users_group_name': users_group_name,
 
                                  'user_group_description': u'DESC',
 
                                  'active': True,
 
                                  '_authentication_token': self.authentication_token()})
 
        response.follow()
 

	
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        self.checkSessionFlash(response,
 
                               'Created user group ')
 
        ## ENABLE REPO CREATE ON A GROUP
 
        response = self.app.put(url('edit_user_group_default_perms',
 
                                    id=ug.users_group_id),
 
                                {'fork_repo_perm': True, '_authentication_token': self.authentication_token()})
 

	
 
        response.follow()
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        p = Permission.get_by_key('hg.create.none')
 
        p2 = Permission.get_by_key('hg.usergroup.create.false')
 
        p3 = Permission.get_by_key('hg.fork.repository')
 
        # check if user has this perms, they should be here since
 
        # defaults are on
 
        perms = UserGroupToPerm.query() \
 
            .filter(UserGroupToPerm.users_group == ug).all()
 

	
 
        assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
 
                    [ug.users_group_id, p2.permission_id],
 
                    [ug.users_group_id, p3.permission_id]])
 

	
 
        ## DISABLE REPO CREATE ON A GROUP
 
        response = self.app.put(url('edit_user_group_default_perms', id=ug.users_group_id),
 
            params={'_authentication_token': self.authentication_token()})
 

	
 
        response.follow()
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        p = Permission.get_by_key('hg.create.none')
 
        p2 = Permission.get_by_key('hg.usergroup.create.false')
 
        p3 = Permission.get_by_key('hg.fork.none')
 
        # check if user has this perms, they should be here since
 
        # defaults are on
 
        perms = UserGroupToPerm.query() \
 
            .filter(UserGroupToPerm.users_group == ug).all()
 

	
 
        assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
 
                    [ug.users_group_id, p2.permission_id],
 
                    [ug.users_group_id, p3.permission_id]])
 

	
 
        # DELETE !
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        ugid = ug.users_group_id
 
        response = self.app.post(url('delete_users_group', id=ug.users_group_id),
 
            params={'_authentication_token': self.authentication_token()})
 
        response = response.follow()
 
        gr = Session().query(UserGroup) \
 
                           .filter(UserGroup.users_group_name ==
 
                                   users_group_name).scalar()
 

	
 
        assert gr == None
 
        p = Permission.get_by_key('hg.fork.repository')
 
        perms = UserGroupToPerm.query() \
 
            .filter(UserGroupToPerm.users_group_id == ugid).all()
 
        perms = [[x.users_group_id,
 
                  x.permission_id, ] for x in perms]
 
        assert perms == []
 

	
 
    def test_delete_browser_fakeout(self):
 
        response = self.app.post(url('delete_users_group', id=1),
 
                                 params=dict(_authentication_token=self.authentication_token()))
0 comments (0 inline, 0 general)