Changeset - 60bf1b171015
Parent rev.
Child rev.
[Not reviewed]
default
0 5 0
Mads Kiilerich - 9 years ago 2016-08-04 14:23:36
madski@unity3d.com
routing: use POST to 'edit_user_ips_update' instead of PUT
5 files changed with 7 insertions and 7 deletions:
kallithea/config/routing.py
2
2
kallithea/templates/admin/permissions/permissions_ips.html
1
1
kallithea/templates/admin/users/user_edit_ips.html
1
1
kallithea/tests/functional/test_admin_permissions.py
1
1
kallithea/tests/functional/test_admin_users.py
2
2
0 comments (0 inline, 0 general)
kallithea/config/routing.py
➞
Show inline comments
 
@@ -20,386 +20,386 @@ refer to the routes manual at http://rou
 
"""
 

			




	
	
	
		
 
from routes import Mapper

			




	
	
	
		
 

			




	
	
	
		
 
# prefix for non repository related links needs to be prefixed with `/`

			




	
	
	
		
 
ADMIN_PREFIX = '/_admin'

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def make_map(config):

			




	
	
	
		
 
    """Create, configure and return the routes Mapper"""

			




	
	
	
		
 
    rmap = Mapper(directory=config['pylons.paths']['controllers'],

			




	
	
	
		
 
                  always_scan=config['debug'])

			




	
	
	
		
 
    rmap.minimization = False

			




	
	
	
		
 
    rmap.explicit = False

			




	
	
	
		
 

			




	
	
	
		
 
    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=["PUT"]))

			




	
	
	
		
 
        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", "/users/{id}/edit/emails",

			




	
	
	
		
 
                  action="add_email", conditions=dict(method=["PUT"]))

			




	
	
	
		
 
        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", "/users/{id}/edit/ips",

			




	
	
	
		
 
                  action="add_ip", conditions=dict(method=["PUT"]))

			




	
	
	
		
 
        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"]))

			




	
	
	
		
 
        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"]))

			




	
	
	
		
 

			




        

    


    
    

    

        

                

                    kallithea/templates/admin/permissions/permissions_ips.html
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 
<div class="ips_wrap">

			




	
	
	
		
 
      <table class="noborder">

			




	
	
	
		
 
      %if c.user_ip_map:

			




	
	
	
		
 
        %for ip in c.user_ip_map:

			




	
	
	
		
 
          <tr>

			




	
	
	
		
 
              <td><div class="ip">${ip.ip_addr}</div></td>

			




	
	
	
		
 
              <td><div class="ip">${h.ip_range(ip.ip_addr)}</div></td>

			




	
	
	
		
 
              <td>

			




	
	
	
		
 
                ${h.form(url('edit_user_ips_delete', id=c.user.user_id))}

			




	
	
	
		
 
                    ${h.hidden('del_ip_id',ip.ip_id)}

			




	
	
	
		
 
                    ${h.hidden('default_user', 'True')}

			




	
	
	
		
 
                    <i class="icon-minus-circled" style="color:#FF4444"></i> ${h.submit('remove_',_('Delete'),id="remove_ip_%s" % ip.ip_id,

			




	
	
	
		
 
                    class_="action_button", onclick="return confirm('"+_('Confirm to delete this IP address: %s') % ip.ip_addr+"');")}

			




	
	
	
		
 
                ${h.end_form()}

			




	
	
	
		
 
              </td>

			




	
	
	
		
 
          </tr>

			




	
	
	
		
 
        %endfor

			




	
	
	
		
 
       %else:

			




	
	
	
		
 
        <tr><td><div class="ip">${_('All IP addresses are allowed.')}</div></td></tr>

			




	
	
	
		
 
       %endif

			




	
	
	
		
 
      </table>

			




	
	
	
		
 
</div>

			




	
	
	
		
 

			




	
	
	
		
 
${h.form(url('edit_user_ips', id=c.user.user_id),method='put')}

			




	
	
	
		
 
${h.form(url('edit_user_ips_update', id=c.user.user_id))}

			




	
	
	
		
 
    <div class="form">

			




	
	
	
		
 
        <!-- fields -->

			




	
	
	
		
 
        <div class="fields">

			




	
	
	
		
 
             <div class="field">

			




	
	
	
		
 
                <div class="label">

			




	
	
	
		
 
                    <label for="new_ip">${_('New IP address')}:</label>

			




	
	
	
		
 
                </div>

			




	
	
	
		
 
                <div class="input">

			




	
	
	
		
 
                    ${h.hidden('default_user', 'True')}

			




	
	
	
		
 
                    ${h.text('new_ip', class_='medium')}

			




	
	
	
		
 
                </div>

			




	
	
	
		
 
             </div>

			




	
	
	
		
 
            <div class="buttons">

			




	
	
	
		
 
              ${h.submit('save',_('Add'),class_="btn")}

			




	
	
	
		
 
              ${h.reset('reset',_('Reset'),class_="btn")}

			




	
	
	
		
 
            </div>

			




	
	
	
		
 
        </div>

			




	
	
	
		
 
    </div>

			




	
	
	
		
 
${h.end_form()}

			




        

    


    
    

    

        

                

                    kallithea/templates/admin/users/user_edit_ips.html
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 
<div class="ips_wrap">

			




	
	
	
		
 
  <table class="noborder">

			




	
	
	
		
 
    %if c.default_user_ip_map and c.inherit_default_ips:

			




	
	
	
		
 
        %for ip in c.default_user_ip_map:

			




	
	
	
		
 
          <tr>

			




	
	
	
		
 
            <td><div class="ip">${ip.ip_addr}</div></td>

			




	
	
	
		
 
            <td><div class="ip">${h.ip_range(ip.ip_addr)}</div></td>

			




	
	
	
		
 
            <td>${h.literal(_('Inherited from %s') % h.link_to('*default*',h.url('admin_permissions_ips')))}</td>

			




	
	
	
		
 
          </tr>

			




	
	
	
		
 
        %endfor

			




	
	
	
		
 
    %endif

			




	
	
	
		
 

			




	
	
	
		
 
    %if c.user_ip_map:

			




	
	
	
		
 
        %for ip in c.user_ip_map:

			




	
	
	
		
 
          <tr>

			




	
	
	
		
 
            <td><div class="ip">${ip.ip_addr}</div></td>

			




	
	
	
		
 
            <td><div class="ip">${h.ip_range(ip.ip_addr)}</div></td>

			




	
	
	
		
 
            <td>

			




	
	
	
		
 
                ${h.form(url('edit_user_ips_delete', id=c.user.user_id))}

			




	
	
	
		
 
                    ${h.hidden('del_ip_id',ip.ip_id)}

			




	
	
	
		
 
                    <i class="icon-minus-circled" style="color:#FF4444"></i>

			




	
	
	
		
 
                    ${h.submit('remove_',_('Delete'),id="remove_ip_%s" % ip.ip_id,

			




	
	
	
		
 
                    class_="action_button", onclick="return  confirm('"+_('Confirm to delete this IP address: %s') % ip.ip_addr+"');")}

			




	
	
	
		
 
                ${h.end_form()}

			




	
	
	
		
 
            </td>

			




	
	
	
		
 
          </tr>

			




	
	
	
		
 
        %endfor

			




	
	
	
		
 
    %endif

			




	
	
	
		
 
    %if not c.default_user_ip_map and not c.user_ip_map:

			




	
	
	
		
 
        <tr><td><div class="ip">${_('All IP addresses are allowed.')}</div></td></tr>

			




	
	
	
		
 
    %endif

			




	
	
	
		
 
  </table>

			




	
	
	
		
 
</div>

			




	
	
	
		
 

			




	
	
	
		
 
<div>

			




	
	
	
		
 
    ${h.form(url('edit_user_ips', id=c.user.user_id),method='put')}

			




	
	
	
		
 
    ${h.form(url('edit_user_ips_update', id=c.user.user_id))}

			




	
	
	
		
 
    <div class="form">

			




	
	
	
		
 
        <!-- fields -->

			




	
	
	
		
 
        <div class="fields">

			




	
	
	
		
 
             <div class="field">

			




	
	
	
		
 
                <div class="label">

			




	
	
	
		
 
                    <label for="new_ip">${_('New IP address')}:</label>

			




	
	
	
		
 
                </div>

			




	
	
	
		
 
                <div class="input">

			




	
	
	
		
 
                    ${h.text('new_ip', class_='medium')}

			




	
	
	
		
 
                </div>

			




	
	
	
		
 
             </div>

			




	
	
	
		
 
            <div class="buttons">

			




	
	
	
		
 
              ${h.submit('save',_('Add'),class_="btn")}

			




	
	
	
		
 
              ${h.reset('reset',_('Reset'),class_="btn")}

			




	
	
	
		
 
            </div>

			




	
	
	
		
 
        </div>

			




	
	
	
		
 
    </div>

			




	
	
	
		
 
    ${h.end_form()}

			




	
	
	
		
 
</div>

			




        

    


    
    

    

        

                

                    kallithea/tests/functional/test_admin_permissions.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 
import time

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.model.db import User, UserIpMap

			




	
	
	
		
 
from kallithea.tests import *

			




	
	
	
		
 

			




	
	
	
		
 
class TestAdminPermissionsController(TestController):

			




	
	
	
		
 

			




	
	
	
		
 
    def test_index(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        response = self.app.get(url('admin_permissions'))

			




	
	
	
		
 
        # Test response...

			




	
	
	
		
 

			




	
	
	
		
 
    def test_index_ips(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        response = self.app.get(url('admin_permissions_ips'))

			




	
	
	
		
 
        # Test response...

			




	
	
	
		
 
        response.mustcontain('All IP addresses are allowed')

			




	
	
	
		
 

			




	
	
	
		
 
    def test_add_ips(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        default_user_id = User.get_default_user().user_id

			




	
	
	
		
 
        response = self.app.put(url('edit_user_ips', id=default_user_id),

			




	
	
	
		
 
        response = self.app.post(url('edit_user_ips_update', id=default_user_id),

			




	
	
	
		
 
                                 params=dict(new_ip='127.0.0.0/24',

			




	
	
	
		
 
                                 _authentication_token=self.authentication_token()))

			




	
	
	
		
 

			




	
	
	
		
 
        # sleep more than beaker.cache.sql_cache_short.expire to expire user cache

			




	
	
	
		
 
        time.sleep(1.5)

			




	
	
	
		
 
        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')

			




	
	
	
		
 

			




	
	
	
		
 
        ## delete

			




	
	
	
		
 
        default_user_id = User.get_default_user().user_id

			




	
	
	
		
 
        del_ip_id = UserIpMap.query().filter(UserIpMap.user_id ==

			




	
	
	
		
 
                                             default_user_id).first().ip_id

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.post(url('edit_user_ips_delete', id=default_user_id),

			




	
	
	
		
 
                                 params=dict(del_ip_id=del_ip_id,

			




	
	
	
		
 
                                             _authentication_token=self.authentication_token()),

			




	
	
	
		
 
                                 extra_environ={'REMOTE_ADDR': '127.0.0.1'})

			




	
	
	
		
 

			




	
	
	
		
 
        # sleep more than beaker.cache.sql_cache_short.expire to expire user cache

			




	
	
	
		
 
        time.sleep(1.5)

			




	
	
	
		
 

			




	
	
	
		
 
        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...

			




        

    


    
    

    

        

                

                    kallithea/tests/functional/test_admin_users.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -212,386 +212,386 @@ class TestAdminUsersController(TestContr

			




	
	
	
		
 
        new_user = Session().query(User) \

			




	
	
	
		
 
            .filter(User.username == username).one()

			




	
	
	
		
 
        response = self.app.post(url('delete_user', id=new_user.user_id),

			




	
	
	
		
 
            params={'_authentication_token': self.authentication_token()})

			




	
	
	
		
 
        self.checkSessionFlash(response, 'User "%s" still '

			




	
	
	
		
 
                               'owns 1 repository groups and cannot be removed. '

			




	
	
	
		
 
                               'Switch owners or remove those repository groups: '

			




	
	
	
		
 
                               '%s' % (username, groupname))

			




	
	
	
		
 

			




	
	
	
		
 
        # Relevant _if_ the user deletion succeeded to make sure we can render groups without owner

			




	
	
	
		
 
        # rg = RepoGroup.get_by_group_name(group_name=groupname)

			




	
	
	
		
 
        # response = self.app.get(url('repos_groups', id=rg.group_id))

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.post(url('delete_repo_group', group_name=groupname),

			




	
	
	
		
 
            params={'_authentication_token': self.authentication_token()})

			




	
	
	
		
 
        self.checkSessionFlash(response, 'Removed repository group %s' % groupname)

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.post(url('delete_user', id=new_user.user_id),

			




	
	
	
		
 
            params={'_authentication_token': self.authentication_token()})

			




	
	
	
		
 
        self.checkSessionFlash(response, 'Successfully deleted user')

			




	
	
	
		
 

			




	
	
	
		
 
    def test_delete_user_group_err(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        username = 'usergrouperr'

			




	
	
	
		
 
        groupname = u'usergroup_fail'

			




	
	
	
		
 

			




	
	
	
		
 
        fixture.create_user(name=username)

			




	
	
	
		
 
        ug = fixture.create_user_group(name=groupname, cur_user=username)

			




	
	
	
		
 

			




	
	
	
		
 
        new_user = Session().query(User) \

			




	
	
	
		
 
            .filter(User.username == username).one()

			




	
	
	
		
 
        response = self.app.post(url('delete_user', id=new_user.user_id),

			




	
	
	
		
 
            params={'_authentication_token': self.authentication_token()})

			




	
	
	
		
 
        self.checkSessionFlash(response, 'User "%s" still '

			




	
	
	
		
 
                               'owns 1 user groups and cannot be removed. '

			




	
	
	
		
 
                               'Switch owners or remove those user groups: '

			




	
	
	
		
 
                               '%s' % (username, groupname))

			




	
	
	
		
 

			




	
	
	
		
 
        # TODO: why do this fail?

			




	
	
	
		
 
        #response = self.app.delete(url('delete_users_group', id=groupname))

			




	
	
	
		
 
        #self.checkSessionFlash(response, 'Removed user group %s' % groupname)

			




	
	
	
		
 

			




	
	
	
		
 
        fixture.destroy_user_group(ug.users_group_id)

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.post(url('delete_user', id=new_user.user_id),

			




	
	
	
		
 
            params={'_authentication_token': self.authentication_token()})

			




	
	
	
		
 
        self.checkSessionFlash(response, 'Successfully deleted user')

			




	
	
	
		
 

			




	
	
	
		
 
    def test_edit(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_by_username(TEST_USER_ADMIN_LOGIN)

			




	
	
	
		
 
        response = self.app.get(url('edit_user', id=user.user_id))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_add_perm_create_repo(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        perm_none = Permission.get_by_key('hg.create.none')

			




	
	
	
		
 
        perm_create = Permission.get_by_key('hg.create.repository')

			




	
	
	
		
 

			




	
	
	
		
 
        user = UserModel().create_or_update(username='dummy', password='qwe',

			




	
	
	
		
 
                                            email='dummy', firstname=u'a',

			




	
	
	
		
 
                                            lastname=u'b')

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 
        uid = user.user_id

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            #User should have None permission on creation repository

			




	
	
	
		
 
            assert UserModel().has_perm(user, perm_none) == False

			




	
	
	
		
 
            assert UserModel().has_perm(user, perm_create) == False

			




	
	
	
		
 

			




	
	
	
		
 
            response = self.app.post(url('edit_user_perms', id=uid),

			




	
	
	
		
 
                                     params=dict(_method='put',

			




	
	
	
		
 
                                                 create_repo_perm=True,

			




	
	
	
		
 
                                                 _authentication_token=self.authentication_token()))

			




	
	
	
		
 

			




	
	
	
		
 
            perm_none = Permission.get_by_key('hg.create.none')

			




	
	
	
		
 
            perm_create = Permission.get_by_key('hg.create.repository')

			




	
	
	
		
 

			




	
	
	
		
 
            #User should have None permission on creation repository

			




	
	
	
		
 
            assert UserModel().has_perm(uid, perm_none) == False

			




	
	
	
		
 
            assert UserModel().has_perm(uid, perm_create) == True

			




	
	
	
		
 
        finally:

			




	
	
	
		
 
            UserModel().delete(uid)

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
    def test_revoke_perm_create_repo(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        perm_none = Permission.get_by_key('hg.create.none')

			




	
	
	
		
 
        perm_create = Permission.get_by_key('hg.create.repository')

			




	
	
	
		
 

			




	
	
	
		
 
        user = UserModel().create_or_update(username='dummy', password='qwe',

			




	
	
	
		
 
                                            email='dummy', firstname=u'a',

			




	
	
	
		
 
                                            lastname=u'b')

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 
        uid = user.user_id

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            #User should have None permission on creation repository

			




	
	
	
		
 
            assert UserModel().has_perm(user, perm_none) == False

			




	
	
	
		
 
            assert UserModel().has_perm(user, perm_create) == False

			




	
	
	
		
 

			




	
	
	
		
 
            response = self.app.post(url('edit_user_perms', id=uid),

			




	
	
	
		
 
                                     params=dict(_method='put', _authentication_token=self.authentication_token()))

			




	
	
	
		
 

			




	
	
	
		
 
            perm_none = Permission.get_by_key('hg.create.none')

			




	
	
	
		
 
            perm_create = Permission.get_by_key('hg.create.repository')

			




	
	
	
		
 

			




	
	
	
		
 
            #User should have None permission on creation repository

			




	
	
	
		
 
            assert UserModel().has_perm(uid, perm_none) == True

			




	
	
	
		
 
            assert UserModel().has_perm(uid, perm_create) == False

			




	
	
	
		
 
        finally:

			




	
	
	
		
 
            UserModel().delete(uid)

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
    def test_add_perm_fork_repo(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        perm_none = Permission.get_by_key('hg.fork.none')

			




	
	
	
		
 
        perm_fork = Permission.get_by_key('hg.fork.repository')

			




	
	
	
		
 

			




	
	
	
		
 
        user = UserModel().create_or_update(username='dummy', password='qwe',

			




	
	
	
		
 
                                            email='dummy', firstname=u'a',

			




	
	
	
		
 
                                            lastname=u'b')

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 
        uid = user.user_id

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            #User should have None permission on creation repository

			




	
	
	
		
 
            assert UserModel().has_perm(user, perm_none) == False

			




	
	
	
		
 
            assert UserModel().has_perm(user, perm_fork) == False

			




	
	
	
		
 

			




	
	
	
		
 
            response = self.app.post(url('edit_user_perms', id=uid),

			




	
	
	
		
 
                                     params=dict(_method='put',

			




	
	
	
		
 
                                                 create_repo_perm=True,

			




	
	
	
		
 
                                                 _authentication_token=self.authentication_token()))

			




	
	
	
		
 

			




	
	
	
		
 
            perm_none = Permission.get_by_key('hg.create.none')

			




	
	
	
		
 
            perm_create = Permission.get_by_key('hg.create.repository')

			




	
	
	
		
 

			




	
	
	
		
 
            #User should have None permission on creation repository

			




	
	
	
		
 
            assert UserModel().has_perm(uid, perm_none) == False

			




	
	
	
		
 
            assert UserModel().has_perm(uid, perm_create) == True

			




	
	
	
		
 
        finally:

			




	
	
	
		
 
            UserModel().delete(uid)

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
    def test_revoke_perm_fork_repo(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        perm_none = Permission.get_by_key('hg.fork.none')

			




	
	
	
		
 
        perm_fork = Permission.get_by_key('hg.fork.repository')

			




	
	
	
		
 

			




	
	
	
		
 
        user = UserModel().create_or_update(username='dummy', password='qwe',

			




	
	
	
		
 
                                            email='dummy', firstname=u'a',

			




	
	
	
		
 
                                            lastname=u'b')

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 
        uid = user.user_id

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            #User should have None permission on creation repository

			




	
	
	
		
 
            assert UserModel().has_perm(user, perm_none) == False

			




	
	
	
		
 
            assert UserModel().has_perm(user, perm_fork) == False

			




	
	
	
		
 

			




	
	
	
		
 
            response = self.app.post(url('edit_user_perms', id=uid),

			




	
	
	
		
 
                                     params=dict(_method='put', _authentication_token=self.authentication_token()))

			




	
	
	
		
 

			




	
	
	
		
 
            perm_none = Permission.get_by_key('hg.create.none')

			




	
	
	
		
 
            perm_create = Permission.get_by_key('hg.create.repository')

			




	
	
	
		
 

			




	
	
	
		
 
            #User should have None permission on creation repository

			




	
	
	
		
 
            assert UserModel().has_perm(uid, perm_none) == True

			




	
	
	
		
 
            assert UserModel().has_perm(uid, perm_create) == False

			




	
	
	
		
 
        finally:

			




	
	
	
		
 
            UserModel().delete(uid)

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ips(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

			




	
	
	
		
 
        response = self.app.get(url('edit_user_ips', id=user.user_id))

			




	
	
	
		
 
        response.mustcontain('All IP addresses are allowed')

			




	
	
	
		
 

			




	
	
	
		
 
    @parametrize('test_name,ip,ip_range,failure', [

			




	
	
	
		
 
        ('127/24', '127.0.0.1/24', '127.0.0.0 - 127.0.0.255', False),

			




	
	
	
		
 
        ('10/32', '10.0.0.10/32', '10.0.0.10 - 10.0.0.10', False),

			




	
	
	
		
 
        ('0/16', '0.0.0.0/16', '0.0.0.0 - 0.0.255.255', False),

			




	
	
	
		
 
        ('0/8', '0.0.0.0/8', '0.0.0.0 - 0.255.255.255', False),

			




	
	
	
		
 
        ('127_bad_mask', '127.0.0.1/99', '127.0.0.1 - 127.0.0.1', True),

			




	
	
	
		
 
        ('127_bad_ip', 'foobar', 'foobar', True),

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_add_ip(self, test_name, ip, ip_range, failure):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

			




	
	
	
		
 
        user_id = user.user_id

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.put(url('edit_user_ips', id=user_id),

			




	
	
	
		
 
                                params=dict(new_ip=ip, _authentication_token=self.authentication_token()))

			




	
	
	
		
 
        response = self.app.post(url('edit_user_ips_update', id=user_id),

			




	
	
	
		
 
                                 params=dict(new_ip=ip, _authentication_token=self.authentication_token()))

			




	
	
	
		
 

			




	
	
	
		
 
        if failure:

			




	
	
	
		
 
            self.checkSessionFlash(response, 'Please enter a valid IPv4 or IPv6 address')

			




	
	
	
		
 
            response = self.app.get(url('edit_user_ips', id=user_id))

			




	
	
	
		
 
            response.mustcontain(no=[ip])

			




	
	
	
		
 
            response.mustcontain(no=[ip_range])

			




	
	
	
		
 

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            response = self.app.get(url('edit_user_ips', id=user_id))

			




	
	
	
		
 
            response.mustcontain(ip)

			




	
	
	
		
 
            response.mustcontain(ip_range)

			




	
	
	
		
 

			




	
	
	
		
 
        ## cleanup

			




	
	
	
		
 
        for del_ip in UserIpMap.query().filter(UserIpMap.user_id == user_id).all():

			




	
	
	
		
 
            Session().delete(del_ip)

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
    def test_delete_ip(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

			




	
	
	
		
 
        user_id = user.user_id

			




	
	
	
		
 
        ip = '127.0.0.1/32'

			




	
	
	
		
 
        ip_range = '127.0.0.1 - 127.0.0.1'

			




	
	
	
		
 
        new_ip = UserModel().add_extra_ip(user_id, ip)

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 
        new_ip_id = new_ip.ip_id

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.get(url('edit_user_ips', id=user_id))

			




	
	
	
		
 
        response.mustcontain(ip)

			




	
	
	
		
 
        response.mustcontain(ip_range)

			




	
	
	
		
 

			




	
	
	
		
 
        self.app.post(url('edit_user_ips_delete', id=user_id),

			




	
	
	
		
 
                      params=dict(del_ip_id=new_ip_id, _authentication_token=self.authentication_token()))

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.get(url('edit_user_ips', id=user_id))

			




	
	
	
		
 
        response.mustcontain('All IP addresses are allowed')

			




	
	
	
		
 
        response.mustcontain(no=[ip])

			




	
	
	
		
 
        response.mustcontain(no=[ip_range])

			




	
	
	
		
 

			




	
	
	
		
 
    def test_api_keys(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 

			




	
	
	
		
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

			




	
	
	
		
 
        response = self.app.get(url('edit_user_api_keys', id=user.user_id))

			




	
	
	
		
 
        response.mustcontain(user.api_key)

			




	
	
	
		
 
        response.mustcontain('Expires: Never')

			




	
	
	
		
 

			




	
	
	
		
 
    @parametrize('desc,lifetime', [

			




	
	
	
		
 
        ('forever', -1),

			




	
	
	
		
 
        ('5mins', 60*5),

			




	
	
	
		
 
        ('30days', 60*60*24*30),

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_add_api_keys(self, desc, lifetime):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

			




	
	
	
		
 
        user_id = user.user_id

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.post(url('edit_user_api_keys', id=user_id),

			




	
	
	
		
 
                 {'description': desc, 'lifetime': lifetime, '_authentication_token': self.authentication_token()})

			




	
	
	
		
 
        self.checkSessionFlash(response, 'API key successfully created')

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            response = response.follow()

			




	
	
	
		
 
            user = User.get(user_id)

			




	
	
	
		
 
            for api_key in user.api_keys:

			




	
	
	
		
 
                response.mustcontain(api_key)

			




	
	
	
		
 
        finally:

			




	
	
	
		
 
            for api_key in UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all():

			




	
	
	
		
 
                Session().delete(api_key)

			




	
	
	
		
 
                Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
    def test_remove_api_key(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

			




	
	
	
		
 
        user_id = user.user_id

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.post(url('edit_user_api_keys', id=user_id),

			




	
	
	
		
 
                {'description': 'desc', 'lifetime': -1, '_authentication_token': self.authentication_token()})

			




	
	
	
		
 
        self.checkSessionFlash(response, 'API key successfully created')

			




	
	
	
		
 
        response = response.follow()

			




	
	
	
		
 

			




	
	
	
		
 
        #now delete our key

			




	
	
	
		
 
        keys = UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all()

			




	
	
	
		
 
        assert 1 == len(keys)

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.post(url('edit_user_api_keys_delete', id=user_id),

			




	
	
	
		
 
                 {'del_api_key': keys[0].api_key, '_authentication_token': self.authentication_token()})

			




	
	
	
		
 
        self.checkSessionFlash(response, 'API key successfully deleted')

			




	
	
	
		
 
        keys = UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all()

			




	
	
	
		
 
        assert 0 == len(keys)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_reset_main_api_key(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

			




	
	
	
		
 
        user_id = user.user_id

			




	
	
	
		
 
        api_key = user.api_key

			




	
	
	
		
 
        response = self.app.get(url('edit_user_api_keys', id=user_id))

			




	
	
	
		
 
        response.mustcontain(api_key)

			




	
	
	
		
 
        response.mustcontain('Expires: Never')

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.post(url('edit_user_api_keys_delete', id=user_id),

			




	
	
	
		
 
                 {'del_api_key_builtin': api_key, '_authentication_token': self.authentication_token()})

			




	
	
	
		
 
        self.checkSessionFlash(response, 'API key successfully reset')

			




	
	
	
		
 
        response = response.follow()

			




	
	
	
		
 
        response.mustcontain(no=[api_key])

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TestAdminUsersController_unittest(object):

			




	
	
	
		
 
    """ Unit tests for the users controller """

			




	
	
	
		
 

			




	
	
	
		
 
    def test_get_user_or_raise_if_default(self, monkeypatch):

			




	
	
	
		
 
        # flash complains about an non-existing session

			




	
	
	
		
 
        def flash_mock(*args, **kwargs):

			




	
	
	
		
 
            pass

			




	
	
	
		
 
        monkeypatch.setattr(h, 'flash', flash_mock)

			




	
	
	
		
 

			




	
	
	
		
 
        u = UsersController()

			




	
	
	
		
 
        # a regular user should work correctly

			




	
	
	
		
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

			




	
	
	
		
 
        assert u._get_user_or_raise_if_default(user.user_id) == user

			




	
	
	
		
 
        # the default user should raise

			




	
	
	
		
 
        with pytest.raises(HTTPNotFound):

			




	
	
	
		
 
            u._get_user_or_raise_if_default(User.get_default_user().user_id)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TestAdminUsersControllerForDefaultUser(TestController):

			




	
	
	
		
 
    """

			




	
	
	
		
 
    Edit actions on the default user are not allowed.

			




	
	
	
		
 
    Validate that they throw a 404 exception.

			




	
	
	
		
 
    """

			




	
	
	
		
 
    def test_edit_default_user(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_default_user()

			




	
	
	
		
 
        response = self.app.get(url('edit_user', id=user.user_id), status=404)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_edit_advanced_default_user(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_default_user()

			




	
	
	
		
 
        response = self.app.get(url('edit_user_advanced', id=user.user_id), status=404)

			




	
	
	
		
 

			




	
	
	
		
 
    # API keys

			




	
	
	
		
 
    def test_edit_api_keys_default_user(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_default_user()

			




	
	
	
		
 
        response = self.app.get(url('edit_user_api_keys', id=user.user_id), status=404)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_add_api_keys_default_user(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_default_user()

			




	
	
	
		
 
        response = self.app.post(url('edit_user_api_keys', id=user.user_id),

			




	
	
	
		
 
                 {'_method': 'put', '_authentication_token': self.authentication_token()}, status=404)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_delete_api_keys_default_user(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_default_user()

			




	
	
	
		
 
        response = self.app.post(url('edit_user_api_keys_delete', id=user.user_id),

			




	
	
	
		
 
                 {'_authentication_token': self.authentication_token()}, status=404)

			




	
	
	
		
 

			




	
	
	
		
 
    # Permissions

			




	
	
	
		
 
    def test_edit_perms_default_user(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_default_user()

			




	
	
	
		
 
        response = self.app.get(url('edit_user_perms', id=user.user_id), status=404)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_update_perms_default_user(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_default_user()

			




	
	
	
		
 
        response = self.app.post(url('edit_user_perms', id=user.user_id),

			




	
	
	
		
 
                 {'_method': 'put', '_authentication_token': self.authentication_token()}, status=404)

			




	
	
	
		
 

			




	
	
	
		
 
    # Emails

			




	
	
	
		
 
    def test_edit_emails_default_user(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_default_user()

			




	
	
	
		
 
        response = self.app.get(url('edit_user_emails', id=user.user_id), status=404)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_add_emails_default_user(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_default_user()

			




	
	
	
		
 
        response = self.app.post(url('edit_user_emails', id=user.user_id),

			




	
	
	
		
 
                 {'_method': 'put', '_authentication_token': self.authentication_token()}, status=404)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_delete_emails_default_user(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        user = User.get_default_user()

			




	
	
	
		
 
        response = self.app.post(url('edit_user_emails_delete', id=user.user_id),

			




	
	
	
		
 
                 {'_authentication_token': self.authentication_token()}, status=404)

			




	
	
	
		
 

			




	
	
	
		
 
    # IP addresses

			




	
	
	
		
 
    # Add/delete of IP addresses for the default user is used to maintain

			




	
	
	
		
 
    # the global IP whitelist and thus allowed. Only 'edit' is forbidden.

			




	
	
	
		
 
    def test_edit_ip_default_user(self):

			




	
	
	
		
 
        self.log_user()

			




        

    



    


    



    



      

      





    
    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.