admin: don't redirect back to the list of users after creating a user

Staying on the user page gives a more natural flow, instead of having to click
the link in the 'Created' flash message to get back to the user.

kallithea/controllers/admin/users.py

@@ -1,446 +1,445 b''
 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (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.users
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 Users 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: Apr 4, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 
 import logging
 import traceback
 import formencode
 
 from formencode import htmlfill
 from pylons import request, tmpl_context as c, url, config
 from pylons.i18n.translation import _
 from sqlalchemy.sql.expression import func
 from webob.exc import HTTPFound, HTTPNotFound
 
 import kallithea
 from kallithea.lib.exceptions import DefaultUserException, \
     UserOwnsReposException, UserCreationError
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import LoginRequired, HasPermissionAnyDecorator, \
     AuthUser
 from kallithea.lib import auth_modules
 from kallithea.lib.auth_modules import auth_internal
 from kallithea.lib.base import BaseController, render
 from kallithea.model.api_key import ApiKeyModel
 
 from kallithea.model.db import User, UserEmailMap, UserIpMap, UserToPerm
 from kallithea.model.forms import UserForm, CustomDefaultPermissionsForm
 from kallithea.model.user import UserModel
 from kallithea.model.meta import Session
 from kallithea.lib.utils import action_logger
 from kallithea.lib.compat import json
 from kallithea.lib.utils2 import datetime_to_time, safe_int, generate_api_key
 
 log = logging.getLogger(__name__)
 
 
 class UsersController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
 
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin')
     def __before__(self):
         super(UsersController, self).__before__()
         c.available_permissions = config['available_permissions']
         c.EXTERN_TYPE_INTERNAL = kallithea.EXTERN_TYPE_INTERNAL
 
     def index(self, format='html'):
         c.users_list = User.query().order_by(User.username) \
                         .filter(User.username != User.DEFAULT_USER) \
                         .order_by(func.lower(User.username)) \
                         .all()
 
         users_data = []
         total_records = len(c.users_list)
         _tmpl_lookup = kallithea.CONFIG['pylons.app_globals'].mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
 
         grav_tmpl = '<div class="gravatar">%s</div>'
 
         username = lambda user_id, username: (
                 template.get_def("user_name")
                 .render(user_id, username, _=_, h=h, c=c))
 
         user_actions = lambda user_id, username: (
                 template.get_def("user_actions")
                 .render(user_id, username, _=_, h=h, c=c))
 
         for user in c.users_list:
             users_data.append({
                 "gravatar": grav_tmpl % h.gravatar(user.email, size=20),
                 "raw_name": user.username,
                 "username": username(user.user_id, user.username),
                 "firstname": h.escape(user.name),
                 "lastname": h.escape(user.lastname),
                 "last_login": h.fmt_date(user.last_login),
                 "last_login_raw": datetime_to_time(user.last_login),
                 "active": h.boolicon(user.active),
                 "admin": h.boolicon(user.admin),
                 "extern_type": user.extern_type,
                 "extern_name": user.extern_name,
                 "action": user_actions(user.user_id, user.username),
             })
 
         c.data = json.dumps({
             "totalRecords": total_records,
             "startIndex": 0,
             "sort": None,
             "dir": "asc",
             "records": users_data
         })
 
         return render('admin/users/users.html')
 
     def create(self):
         c.default_extern_type = auth_internal.KallitheaAuthPlugin.name
         c.default_extern_name = auth_internal.KallitheaAuthPlugin.name
         user_model = UserModel()
         user_form = UserForm()()
         try:
             form_result = user_form.to_python(dict(request.POST))
             user = user_model.create(form_result)
-            usr = form_result['username']
-            action_logger(self.authuser, 'admin_created_user:%s' % usr,
+            action_logger(self.authuser, 'admin_created_user:%s' % user.username,
                           None, self.ip_addr, self.sa)
-            h.flash(h.literal(_('Created user %s') % h.link_to(h.escape(usr), url('edit_user', id=user.user_id))),
+            h.flash(_('Created user %s') % user.username,
                     category='success')
             Session().commit()
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('admin/users/user_add.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except UserCreationError as e:
             h.flash(e, 'error')
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('Error occurred during creation of user %s') \
                     % request.POST.get('username'), category='error')
-        raise HTTPFound(location=url('users'))
+        raise HTTPFound(location=url('edit_user', id=user.user_id))
 
     def new(self, format='html'):
         c.default_extern_type = auth_internal.KallitheaAuthPlugin.name
         c.default_extern_name = auth_internal.KallitheaAuthPlugin.name
         return render('admin/users/user_add.html')
 
     def update(self, id):
         user_model = UserModel()
         user = user_model.get(id)
         _form = UserForm(edit=True, old_data={'user_id': id,
                                               'email': user.email})()
         form_result = {}
         try:
             form_result = _form.to_python(dict(request.POST))
             skip_attrs = ['extern_type', 'extern_name',
                          ] + auth_modules.get_managed_fields(user)
 
             user_model.update(id, form_result, skip_attrs=skip_attrs)
             usr = form_result['username']
             action_logger(self.authuser, 'admin_updated_user:%s' % usr,
                           None, self.ip_addr, self.sa)
             h.flash(_('User updated successfully'), category='success')
             Session().commit()
         except formencode.Invalid as errors:
             defaults = errors.value
             e = errors.error_dict or {}
             defaults.update({
                 'create_repo_perm': user_model.has_perm(id,
                                                         'hg.create.repository'),
                 'fork_repo_perm': user_model.has_perm(id, 'hg.fork.repository'),
             })
             return htmlfill.render(
                 self._render_edit_profile(user),
                 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 %s') \
                     % form_result.get('username'), category='error')
         raise HTTPFound(location=url('edit_user', id=id))
 
     def delete(self, id):
         usr = User.get_or_404(id)
         try:
             UserModel().delete(usr)
             Session().commit()
             h.flash(_('Successfully deleted user'), category='success')
         except (UserOwnsReposException, DefaultUserException) as e:
             h.flash(e, category='warning')
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during deletion of user'),
                     category='error')
         raise HTTPFound(location=url('users'))
 
     def _get_user_or_raise_if_default(self, id):
         try:
             return User.get_or_404(id, allow_default=False)
         except DefaultUserException:
             h.flash(_("The default user cannot be edited"), category='warning')
             raise HTTPNotFound
 
     def _render_edit_profile(self, user):
         c.user = user
         c.active = 'profile'
         c.perm_user = AuthUser(dbuser=user)
         c.ip_addr = self.ip_addr
         managed_fields = auth_modules.get_managed_fields(user)
         c.readonly = lambda n: 'readonly' if n in managed_fields else None
         return render('admin/users/user_edit.html')
 
     def edit(self, id, format='html'):
         user = self._get_user_or_raise_if_default(id)
         defaults = user.get_dict()
 
         return htmlfill.render(
             self._render_edit_profile(user),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
 
     def edit_advanced(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         c.active = 'advanced'
         c.perm_user = AuthUser(user_id=id)
         c.ip_addr = self.ip_addr
 
         umodel = UserModel()
         defaults = c.user.get_dict()
         defaults.update({
             'create_repo_perm': umodel.has_perm(c.user, 'hg.create.repository'),
             'create_user_group_perm': umodel.has_perm(c.user,
                                                       'hg.usergroup.create.true'),
             'fork_repo_perm': umodel.has_perm(c.user, 'hg.fork.repository'),
         })
         return htmlfill.render(
             render('admin/users/user_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
 
     def edit_api_keys(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         c.active = 'api_keys'
         show_expired = True
         c.lifetime_values = [
             (str(-1), _('Forever')),
             (str(5), _('5 minutes')),
             (str(60), _('1 hour')),
             (str(60 * 24), _('1 day')),
             (str(60 * 24 * 30), _('1 month')),
         ]
         c.lifetime_options = [(c.lifetime_values, _("Lifetime"))]
         c.user_api_keys = ApiKeyModel().get_api_keys(c.user.user_id,
                                                      show_expired=show_expired)
         defaults = c.user.get_dict()
         return htmlfill.render(
             render('admin/users/user_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
 
     def add_api_key(self, id):
         c.user = self._get_user_or_raise_if_default(id)
 
         lifetime = safe_int(request.POST.get('lifetime'), -1)
         description = request.POST.get('description')
         ApiKeyModel().create(c.user.user_id, description, lifetime)
         Session().commit()
         h.flash(_("API key successfully created"), category='success')
         raise HTTPFound(location=url('edit_user_api_keys', id=c.user.user_id))
 
     def delete_api_key(self, id):
         c.user = self._get_user_or_raise_if_default(id)
 
         api_key = request.POST.get('del_api_key')
         if request.POST.get('del_api_key_builtin'):
             user = User.get(c.user.user_id)
             if user is not None:
                 user.api_key = generate_api_key()
                 Session().add(user)
                 Session().commit()
                 h.flash(_("API key successfully reset"), category='success')
         elif api_key:
             ApiKeyModel().delete(api_key, c.user.user_id)
             Session().commit()
             h.flash(_("API key successfully deleted"), category='success')
 
         raise HTTPFound(location=url('edit_user_api_keys', id=c.user.user_id))
 
     def update_account(self, id):
         pass
 
     def edit_perms(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         c.active = 'perms'
         c.perm_user = AuthUser(user_id=id)
         c.ip_addr = self.ip_addr
 
         umodel = UserModel()
         defaults = c.user.get_dict()
         defaults.update({
             'create_repo_perm': umodel.has_perm(c.user, 'hg.create.repository'),
             'create_user_group_perm': umodel.has_perm(c.user,
                                                       'hg.usergroup.create.true'),
             'fork_repo_perm': umodel.has_perm(c.user, 'hg.fork.repository'),
         })
         return htmlfill.render(
             render('admin/users/user_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
 
     def update_perms(self, id):
         user = self._get_user_or_raise_if_default(id)
 
         try:
             form = CustomDefaultPermissionsForm()()
             form_result = form.to_python(request.POST)
 
             inherit_perms = form_result['inherit_default_permissions']
             user.inherit_default_permissions = inherit_perms
             Session().add(user)
             user_model = UserModel()
 
             defs = UserToPerm.query() \
                 .filter(UserToPerm.user == user) \
                 .all()
             for ug in defs:
                 Session().delete(ug)
 
             if form_result['create_repo_perm']:
                 user_model.grant_perm(id, 'hg.create.repository')
             else:
                 user_model.grant_perm(id, 'hg.create.none')
             if form_result['create_user_group_perm']:
                 user_model.grant_perm(id, 'hg.usergroup.create.true')
             else:
                 user_model.grant_perm(id, 'hg.usergroup.create.false')
             if form_result['fork_repo_perm']:
                 user_model.grant_perm(id, 'hg.fork.repository')
             else:
                 user_model.grant_perm(id, 'hg.fork.none')
             h.flash(_("Updated permissions"), category='success')
             Session().commit()
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during permissions saving'),
                     category='error')
         raise HTTPFound(location=url('edit_user_perms', id=id))
 
     def edit_emails(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         c.active = 'emails'
         c.user_email_map = UserEmailMap.query() \
             .filter(UserEmailMap.user == c.user).all()
 
         defaults = c.user.get_dict()
         return htmlfill.render(
             render('admin/users/user_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
 
     def add_email(self, id):
         user = self._get_user_or_raise_if_default(id)
         email = request.POST.get('new_email')
         user_model = UserModel()
 
         try:
             user_model.add_extra_email(id, email)
             Session().commit()
             h.flash(_("Added email %s to user") % email, category='success')
         except formencode.Invalid as error:
             msg = error.error_dict['email']
             h.flash(msg, category='error')
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during email saving'),
                     category='error')
         raise HTTPFound(location=url('edit_user_emails', id=id))
 
     def delete_email(self, id):
         user = self._get_user_or_raise_if_default(id)
         email_id = request.POST.get('del_email_id')
         user_model = UserModel()
         user_model.delete_extra_email(id, email_id)
         Session().commit()
         h.flash(_("Removed email from user"), category='success')
         raise HTTPFound(location=url('edit_user_emails', id=id))
 
     def edit_ips(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         c.active = 'ips'
         c.user_ip_map = UserIpMap.query() \
             .filter(UserIpMap.user == c.user).all()
 
         c.inherit_default_ips = c.user.inherit_default_permissions
         c.default_user_ip_map = UserIpMap.query() \
             .filter(UserIpMap.user == User.get_default_user()).all()
 
         defaults = c.user.get_dict()
         return htmlfill.render(
             render('admin/users/user_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
 
     def add_ip(self, id):
         ip = request.POST.get('new_ip')
         user_model = UserModel()
 
         try:
             user_model.add_extra_ip(id, ip)
             Session().commit()
             h.flash(_("Added IP address %s to user whitelist") % ip, category='success')
         except formencode.Invalid as error:
             msg = error.error_dict['ip']
             h.flash(msg, category='error')
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred while adding IP address'),
                     category='error')
 
         if 'default_user' in request.POST:
             raise HTTPFound(location=url('admin_permissions_ips'))
         raise HTTPFound(location=url('edit_user_ips', id=id))
 
     def delete_ip(self, id):
         ip_id = request.POST.get('del_ip_id')
         user_model = UserModel()
         user_model.delete_extra_ip(id, ip_id)
         Session().commit()
         h.flash(_("Removed IP address from user whitelist"), category='success')
 
         if 'default_user' in request.POST:
             raise HTTPFound(location=url('admin_permissions_ips'))
         raise HTTPFound(location=url('edit_user_ips', id=id))

kallithea/tests/functional/test_admin_users.py

@@ -1,602 +1,602 b''
 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (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/>.
 
 from sqlalchemy.orm.exc import NoResultFound, ObjectDeletedError
 
 import pytest
 from kallithea.tests import *
 from kallithea.tests.fixture import Fixture
 from kallithea.controllers.admin.users import UsersController
 from kallithea.model.db import User, Permission, UserIpMap, UserApiKeys
 from kallithea.lib.auth import check_password
 from kallithea.model.user import UserModel
 from kallithea.model import validators
 from kallithea.lib import helpers as h
 from kallithea.model.meta import Session
 from webob.exc import HTTPNotFound
 
 fixture = Fixture()
 
 @pytest.yield_fixture
 def user_and_repo_group_fail():
     username = 'repogrouperr'
     groupname = u'repogroup_fail'
     user = fixture.create_user(name=username)
     repo_group = fixture.create_repo_group(name=groupname, cur_user=username)
     yield user, repo_group
     # cleanup
     try:
         fixture.destroy_repo_group(repo_group)
     except ObjectDeletedError:
         # delete already succeeded in test body
         pass
 
 class TestAdminUsersController(TestController):
     test_user_1 = 'testme'
 
     @classmethod
     def teardown_class(cls):
         if User.get_by_username(cls.test_user_1):
             UserModel().delete(cls.test_user_1)
             Session().commit()
 
     def test_index(self):
         self.log_user()
         response = self.app.get(url('users'))
         # Test response...
 
     def test_create(self):
         self.log_user()
         username = 'newtestuser'
         password = 'test12'
         password_confirmation = password
         name = u'name'
         lastname = u'lastname'
         email = '[e-mail]'
 
         response = self.app.post(url('users'),
             {'username': username,
              'password': password,
              'password_confirmation': password_confirmation,
              'firstname': name,
              'active': True,
              'lastname': lastname,
              'extern_name': 'internal',
              'extern_type': 'internal',
              'email': email,
              '_authentication_token': self.authentication_token()})
+        # 302 Found
+        # The resource was found at http://localhost/_admin/users/5/edit; you should be redirected automatically.
 
-        self.checkSessionFlash(response, '''Created user <a href="/_admin/users/''')
-        self.checkSessionFlash(response, '''/edit">%s</a>''' % (username))
+        self.checkSessionFlash(response, '''Created user %s''' % username)
+
+        response = response.follow()
+        response.mustcontain("""%s user settings""" % username) # in <title>
 
         new_user = Session().query(User). \
             filter(User.username == username).one()
 
         assert new_user.username == username
         assert check_password(password, new_user.password) == True
         assert new_user.name == name
         assert new_user.lastname == lastname
         assert new_user.email == email
 
-        response.follow()
-        response = response.follow()
-        response.mustcontain("""newtestuser""")
-
     def test_create_err(self):
         self.log_user()
         username = 'new_user'
         password = ''
         name = u'name'
         lastname = u'lastname'
         email = 'errmail.example.com'
 
         response = self.app.post(url('users'), {'username': username,
                                                'password': password,
                                                'name': name,
                                                'active': False,
                                                'lastname': lastname,
                                                'email': email,
                                                '_authentication_token': self.authentication_token()})
 
         msg = validators.ValidUsername(False, {})._messages['system_invalid_username']
         msg = h.html_escape(msg % {'username': 'new_user'})
         response.mustcontain("""<span class="error-message">%s</span>""" % msg)
         response.mustcontain("""<span class="error-message">Please enter a value</span>""")
         response.mustcontain("""<span class="error-message">An email address must contain a single @</span>""")
 
         def get_user():
             Session().query(User).filter(User.username == username).one()
 
         with pytest.raises(NoResultFound):
             get_user(), 'found user in database'
 
     def test_new(self):
         self.log_user()
         response = self.app.get(url('new_user'))
 
     @parametrize('name,attrs',
         [('firstname', {'firstname': 'new_username'}),
          ('lastname', {'lastname': 'new_username'}),
          ('admin', {'admin': True}),
          ('admin', {'admin': False}),
          ('extern_type', {'extern_type': 'ldap'}),
          ('extern_type', {'extern_type': None}),
          ('extern_name', {'extern_name': 'test'}),
          ('extern_name', {'extern_name': None}),
          ('active', {'active': False}),
          ('active', {'active': True}),
          ('email', {'email': '[e-mail]'}),
         # ('new_password', {'new_password': 'foobar123',
         #                   'password_confirmation': 'foobar123'})
         ])
     def test_update(self, name, attrs):
         self.log_user()
         usr = fixture.create_user(self.test_user_1, password='qweqwe',
                                   email='[e-mail]',
                                   extern_type='internal',
                                   extern_name=self.test_user_1,
                                   skip_if_exists=True)
         Session().commit()
         params = usr.get_api_data(True)
         params.update({'password_confirmation': ''})
         params.update({'new_password': ''})
         params.update(attrs)
         if name == 'email':
             params['emails'] = [attrs['email']]
         if name == 'extern_type':
             #cannot update this via form, expected value is original one
             params['extern_type'] = "internal"
         if name == 'extern_name':
             #cannot update this via form, expected value is original one
             params['extern_name'] = self.test_user_1
             # special case since this user is not
                                           # logged in yet his data is not filled
                                           # so we use creation data
 
         params.update({'_authentication_token': self.authentication_token()})
         response = self.app.post(url('update_user', id=usr.user_id), params)
         self.checkSessionFlash(response, 'User updated successfully')
         params.pop('_authentication_token')
 
         updated_user = User.get_by_username(self.test_user_1)
         updated_params = updated_user.get_api_data(True)
         updated_params.update({'password_confirmation': ''})
         updated_params.update({'new_password': ''})
 
         assert params == updated_params
 
     def test_delete(self):
         self.log_user()
         username = 'newtestuserdeleteme'
 
         fixture.create_user(name=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, 'Successfully deleted user')
 
     def test_delete_repo_err(self):
         self.log_user()
         username = 'repoerr'
         reponame = u'repoerr_fail'
 
         fixture.create_user(name=username)
         fixture.create_repo(name=reponame, 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 repositories and cannot be removed. '
                                'Switch owners or remove those repositories: '
                                '%s' % (username, reponame))
 
         response = self.app.post(url('delete_repo', repo_name=reponame),
             params={'_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Deleted repository %s' % reponame)
 
         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_repo_group_err(self, user_and_repo_group_fail):
         self.log_user()
         username = 'repogrouperr'
         groupname = u'repogroup_fail'
 
         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_update', id=uid),
                                      params=dict(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_update', id=uid),
                                      params=dict(_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_update', id=uid),
                                      params=dict(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_update', id=uid),
                                      params=dict(_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, auto_clear_ip_permissions):
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         user_id = user.user_id
 
         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)
 
     def test_delete_ip(self, auto_clear_ip_permissions):
         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_update', 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_update', 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_update', id=user.user_id),
                  {'_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_update', id=user.user_id),
                  {'_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_update', id=user.user_id),
                  {'_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()
         user = User.get_default_user()
         response = self.app.get(url('edit_user_ips', id=user.user_id), status=404)