tests: made few tests handle database transactions better.

rhodecode/apps/admin/tests/test_admin_main_views.py

@@ -1,82 +1,82 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2018 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import pytest
 
 from rhodecode.tests import TestController
 from rhodecode.tests.fixture import Fixture
 
 fixture = Fixture()
 
 
 def route_path(name, params=None, **kwargs):
     import urllib
     from rhodecode.apps._base import ADMIN_PREFIX
 
     base_url = {
         'admin_home': ADMIN_PREFIX,
         'pullrequest_show': '/{repo_name}/pull-request/{pull_request_id}',
         'pull_requests_global': ADMIN_PREFIX + '/pull-request/{pull_request_id}',
         'pull_requests_global_0': ADMIN_PREFIX + '/pull_requests/{pull_request_id}',
         'pull_requests_global_1': ADMIN_PREFIX + '/pull-requests/{pull_request_id}',
 
     }[name].format(**kwargs)
 
     if params:
         base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
     return base_url
 
 
 class TestAdminMainView(TestController):
 
     def test_redirect_admin_home(self):
         self.log_user()
         response = self.app.get(route_path('admin_home'), status=302)
         assert response.location.endswith('/audit_logs')
 
     def test_redirect_pull_request_view(self, view):
         self.log_user()
         self.app.get(
             route_path(view, pull_request_id='xxxx'),
             status=404)
 
     @pytest.mark.backends("git", "hg")
     @pytest.mark.parametrize('view', [
         'pull_requests_global',
         'pull_requests_global_0',
         'pull_requests_global_1',
     ])
     def test_redirect_pull_request_view(self, view, pr_util):
         self.log_user()
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
+        repo_name = pull_request.target_repo.repo_name
 
         response = self.app.get(
             route_path(view, pull_request_id=pull_request_id),
             status=302)
         assert response.location.endswith(
             'pull-request/{}'.format(pull_request_id))
 
-        repo_name = pull_request.target_repo.repo_name
         redirect_url = route_path(
             'pullrequest_show', repo_name=repo_name,
-            pull_request_id=pull_request.pull_request_id)
+            pull_request_id=pull_request_id)
 
         assert redirect_url in response.location

rhodecode/apps/admin/tests/test_admin_repos.py

@@ -1,509 +1,512 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2018 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import urllib
 
 import mock
 import pytest
 
 from rhodecode.apps._base import ADMIN_PREFIX
 from rhodecode.lib import auth
 from rhodecode.lib.utils2 import safe_str
 from rhodecode.lib import helpers as h
 from rhodecode.model.db import (
     Repository, RepoGroup, UserRepoToPerm, User, Permission)
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.user import UserModel
 from rhodecode.tests import (
     login_user_session, assert_session_flash, TEST_USER_ADMIN_LOGIN,
     TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
 from rhodecode.tests.fixture import Fixture, error_function
 from rhodecode.tests.utils import AssertResponse, repo_on_filesystem
 
 fixture = Fixture()
 
 
 def route_path(name, params=None, **kwargs):
     import urllib
 
     base_url = {
         'repos': ADMIN_PREFIX + '/repos',
         'repo_new': ADMIN_PREFIX + '/repos/new',
         'repo_create': ADMIN_PREFIX + '/repos/create',
 
         'repo_creating_check': '/{repo_name}/repo_creating_check',
     }[name].format(**kwargs)
 
     if params:
         base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
     return base_url
 
 
 def _get_permission_for_user(user, repo):
     perm = UserRepoToPerm.query()\
         .filter(UserRepoToPerm.repository ==
                 Repository.get_by_repo_name(repo))\
         .filter(UserRepoToPerm.user == User.get_by_username(user))\
         .all()
     return perm
 
 
 @pytest.mark.usefixtures("app")
 class TestAdminRepos(object):
 
     def test_repo_list(self, autologin_user, user_util):
         repo = user_util.create_repo()
+        repo_name = repo.repo_name
         response = self.app.get(
             route_path('repos'), status=200)
 
-        response.mustcontain(repo.repo_name)
+        response.mustcontain(repo_name)
 
     def test_create_page_restricted_to_single_backend(self, autologin_user, backend):
         with mock.patch('rhodecode.BACKENDS', {'git': 'git'}):
             response = self.app.get(route_path('repo_new'), status=200)
         assert_response = AssertResponse(response)
         element = assert_response.get_element('#repo_type')
         assert element.text_content() == '\ngit\n'
 
     def test_create_page_non_restricted_backends(self, autologin_user, backend):
         response = self.app.get(route_path('repo_new'), status=200)
         assert_response = AssertResponse(response)
         assert_response.element_contains('#repo_type', 'git')
         assert_response.element_contains('#repo_type', 'svn')
         assert_response.element_contains('#repo_type', 'hg')
 
     @pytest.mark.parametrize(
         "suffix", [u'', u'xxa'], ids=['', 'non-ascii'])
     def test_create(self, autologin_user, backend, suffix, csrf_token):
         repo_name_unicode = backend.new_repo_name(suffix=suffix)
         repo_name = repo_name_unicode.encode('utf8')
         description_unicode = u'description for newly created repo' + suffix
         description = description_unicode.encode('utf8')
         response = self.app.post(
             route_path('repo_create'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 csrf_token=csrf_token),
             status=302)
 
         self.assert_repository_is_created_correctly(
             repo_name, description, backend)
 
     def test_create_numeric_name(self, autologin_user, backend, csrf_token):
         numeric_repo = '1234'
         repo_name = numeric_repo
         description = 'description for newly created repo' + numeric_repo
         self.app.post(
             route_path('repo_create'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 csrf_token=csrf_token))
 
         self.assert_repository_is_created_correctly(
             repo_name, description, backend)
 
     @pytest.mark.parametrize("suffix", [u'', u'ąćę'], ids=['', 'non-ascii'])
     def test_create_in_group(
             self, autologin_user, backend, suffix, csrf_token):
         # create GROUP
         group_name = 'sometest_%s' % backend.alias
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
 
         repo_name = u'ingroup' + suffix
         repo_name_full = RepoGroup.url_sep().join(
             [group_name, repo_name])
         description = u'description for newly created repo'
         self.app.post(
             route_path('repo_create'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=safe_str(repo_name),
                 repo_type=backend.alias,
                 repo_description=description,
                 repo_group=gr.group_id,
                 csrf_token=csrf_token))
 
         # TODO: johbo: Cleanup work to fixture
         try:
             self.assert_repository_is_created_correctly(
                 repo_name_full, description, backend)
 
             new_repo = RepoModel().get_by_repo_name(repo_name_full)
             inherited_perms = UserRepoToPerm.query().filter(
                 UserRepoToPerm.repository_id == new_repo.repo_id).all()
             assert len(inherited_perms) == 1
         finally:
             RepoModel().delete(repo_name_full)
             RepoGroupModel().delete(group_name)
             Session().commit()
 
     def test_create_in_group_numeric_name(
             self, autologin_user, backend, csrf_token):
         # create GROUP
         group_name = 'sometest_%s' % backend.alias
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
 
         repo_name = '12345'
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = 'description for newly created repo'
         self.app.post(
             route_path('repo_create'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 repo_group=gr.group_id,
                 csrf_token=csrf_token))
 
         # TODO: johbo: Cleanup work to fixture
         try:
             self.assert_repository_is_created_correctly(
                 repo_name_full, description, backend)
 
             new_repo = RepoModel().get_by_repo_name(repo_name_full)
             inherited_perms = UserRepoToPerm.query()\
                 .filter(UserRepoToPerm.repository_id == new_repo.repo_id).all()
             assert len(inherited_perms) == 1
         finally:
             RepoModel().delete(repo_name_full)
             RepoGroupModel().delete(group_name)
             Session().commit()
 
     def test_create_in_group_without_needed_permissions(self, backend):
         session = login_user_session(
             self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         csrf_token = auth.get_csrf_token(session)
         # revoke
         user_model = UserModel()
         # disable fork and create on default user
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
 
         # disable on regular user
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
         Session().commit()
 
         # create GROUP
         group_name = 'reg_sometest_%s' % backend.alias
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
+        repo_group_id = gr.group_id
 
         group_name_allowed = 'reg_sometest_allowed_%s' % backend.alias
         gr_allowed = RepoGroupModel().create(
             group_name=group_name_allowed,
             group_description='test',
             owner=TEST_USER_REGULAR_LOGIN)
+        allowed_repo_group_id = gr_allowed.group_id
         Session().commit()
 
         repo_name = 'ingroup'
         description = 'description for newly created repo'
         response = self.app.post(
             route_path('repo_create'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
-                repo_group=gr.group_id,
+                repo_group=repo_group_id,
                 csrf_token=csrf_token))
 
         response.mustcontain('Invalid value')
 
         # user is allowed to create in this group
         repo_name = 'ingroup'
         repo_name_full = RepoGroup.url_sep().join(
             [group_name_allowed, repo_name])
         description = 'description for newly created repo'
         response = self.app.post(
             route_path('repo_create'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
-                repo_group=gr_allowed.group_id,
+                repo_group=allowed_repo_group_id,
                 csrf_token=csrf_token))
 
         # TODO: johbo: Cleanup in pytest fixture
         try:
             self.assert_repository_is_created_correctly(
                 repo_name_full, description, backend)
 
             new_repo = RepoModel().get_by_repo_name(repo_name_full)
             inherited_perms = UserRepoToPerm.query().filter(
                 UserRepoToPerm.repository_id == new_repo.repo_id).all()
             assert len(inherited_perms) == 1
 
             assert repo_on_filesystem(repo_name_full)
         finally:
             RepoModel().delete(repo_name_full)
             RepoGroupModel().delete(group_name)
             RepoGroupModel().delete(group_name_allowed)
             Session().commit()
 
     def test_create_in_group_inherit_permissions(self, autologin_user, backend,
                                                  csrf_token):
         # create GROUP
         group_name = 'sometest_%s' % backend.alias
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         perm = Permission.get_by_key('repository.write')
         RepoGroupModel().grant_user_permission(
             gr, TEST_USER_REGULAR_LOGIN, perm)
 
         # add repo permissions
         Session().commit()
-
+        repo_group_id = gr.group_id
         repo_name = 'ingroup_inherited_%s' % backend.alias
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = 'description for newly created repo'
         self.app.post(
             route_path('repo_create'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
-                repo_group=gr.group_id,
+                repo_group=repo_group_id,
                 repo_copy_permissions=True,
                 csrf_token=csrf_token))
 
         # TODO: johbo: Cleanup to pytest fixture
         try:
             self.assert_repository_is_created_correctly(
                 repo_name_full, description, backend)
         except Exception:
             RepoGroupModel().delete(group_name)
             Session().commit()
             raise
 
         # check if inherited permissions are applied
         new_repo = RepoModel().get_by_repo_name(repo_name_full)
         inherited_perms = UserRepoToPerm.query().filter(
             UserRepoToPerm.repository_id == new_repo.repo_id).all()
         assert len(inherited_perms) == 2
 
         assert TEST_USER_REGULAR_LOGIN in [
             x.user.username for x in inherited_perms]
         assert 'repository.write' in [
             x.permission.permission_name for x in inherited_perms]
 
         RepoModel().delete(repo_name_full)
         RepoGroupModel().delete(group_name)
         Session().commit()
 
     @pytest.mark.xfail_backends(
         "git", "hg", reason="Missing reposerver support")
     def test_create_with_clone_uri(self, autologin_user, backend, reposerver,
                                    csrf_token):
         source_repo = backend.create_repo(number_of_commits=2)
         source_repo_name = source_repo.repo_name
         reposerver.serve(source_repo.scm_instance())
 
         repo_name = backend.new_repo_name()
         response = self.app.post(
             route_path('repo_create'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description='',
                 clone_uri=reposerver.url,
                 csrf_token=csrf_token),
             status=302)
 
         # Should be redirected to the creating page
         response.mustcontain('repo_creating')
 
         # Expecting that both repositories have same history
         source_repo = RepoModel().get_by_repo_name(source_repo_name)
         source_vcs = source_repo.scm_instance()
         repo = RepoModel().get_by_repo_name(repo_name)
         repo_vcs = repo.scm_instance()
         assert source_vcs[0].message == repo_vcs[0].message
         assert source_vcs.count() == repo_vcs.count()
         assert source_vcs.commit_ids == repo_vcs.commit_ids
 
     @pytest.mark.xfail_backends("svn", reason="Depends on import support")
     def test_create_remote_repo_wrong_clone_uri(self, autologin_user, backend,
                                                 csrf_token):
         repo_name = backend.new_repo_name()
         description = 'description for newly created repo'
         response = self.app.post(
             route_path('repo_create'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 clone_uri='http://repo.invalid/repo',
                 csrf_token=csrf_token))
         response.mustcontain('invalid clone url')
 
     @pytest.mark.xfail_backends("svn", reason="Depends on import support")
     def test_create_remote_repo_wrong_clone_uri_hg_svn(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.new_repo_name()
         description = 'description for newly created repo'
         response = self.app.post(
             route_path('repo_create'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 clone_uri='svn+http://svn.invalid/repo',
                 csrf_token=csrf_token))
         response.mustcontain('invalid clone url')
 
     def test_create_with_git_suffix(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.new_repo_name() + ".git"
         description = 'description for newly created repo'
         response = self.app.post(
             route_path('repo_create'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 csrf_token=csrf_token))
         response.mustcontain('Repository name cannot end with .git')
 
     def test_default_user_cannot_access_private_repo_in_a_group(
             self, autologin_user, user_util, backend):
 
         group = user_util.create_repo_group()
 
         repo = backend.create_repo(
             repo_private=True, repo_group=group, repo_copy_permissions=True)
 
         permissions = _get_permission_for_user(
             user='default', repo=repo.repo_name)
         assert len(permissions) == 1
         assert permissions[0].permission.permission_name == 'repository.none'
         assert permissions[0].repository.private is True
 
     def test_create_on_top_level_without_permissions(self, backend):
         session = login_user_session(
             self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         csrf_token = auth.get_csrf_token(session)
 
         # revoke
         user_model = UserModel()
         # disable fork and create on default user
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
 
         # disable on regular user
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
         Session().commit()
 
         repo_name = backend.new_repo_name()
         description = 'description for newly created repo'
         response = self.app.post(
             route_path('repo_create'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 csrf_token=csrf_token))
 
         response.mustcontain(
             u"You do not have the permission to store repositories in "
             u"the root location.")
 
     @mock.patch.object(RepoModel, '_create_filesystem_repo', error_function)
     def test_create_repo_when_filesystem_op_fails(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.new_repo_name()
         description = 'description for newly created repo'
 
         response = self.app.post(
             route_path('repo_create'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 csrf_token=csrf_token))
 
         assert_session_flash(
             response, 'Error creating repository %s' % repo_name)
         # repo must not be in db
         assert backend.repo is None
         # repo must not be in filesystem !
         assert not repo_on_filesystem(repo_name)
 
     def assert_repository_is_created_correctly(
             self, repo_name, description, backend):
         repo_name_utf8 = safe_str(repo_name)
 
         # run the check page that triggers the flash message
         response = self.app.get(
             route_path('repo_creating_check', repo_name=safe_str(repo_name)))
         assert response.json == {u'result': True}
 
         flash_msg = u'Created repository <a href="/{}">{}</a>'.format(
             urllib.quote(repo_name_utf8), repo_name)
         assert_session_flash(response, flash_msg)
 
         # test if the repo was created in the database
         new_repo = RepoModel().get_by_repo_name(repo_name)
 
         assert new_repo.repo_name == repo_name
         assert new_repo.description == description
 
         # test if the repository is visible in the list ?
         response = self.app.get(
             h.route_path('repo_summary', repo_name=safe_str(repo_name)))
         response.mustcontain(repo_name)
         response.mustcontain(backend.alias)
 
         assert repo_on_filesystem(repo_name)

rhodecode/apps/admin/tests/test_admin_user_groups.py

@@ -1,170 +1,170 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2018 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import pytest
 
 from rhodecode.model.db import UserGroup, User
 from rhodecode.model.meta import Session
 
 from rhodecode.tests import (
     TestController, TEST_USER_REGULAR_LOGIN, assert_session_flash)
 from rhodecode.tests.fixture import Fixture
 
 fixture = Fixture()
 
 
 def route_path(name, params=None, **kwargs):
     import urllib
     from rhodecode.apps._base import ADMIN_PREFIX
 
     base_url = {
         'user_groups': ADMIN_PREFIX + '/user_groups',
         'user_groups_data': ADMIN_PREFIX + '/user_groups_data',
         'user_group_members_data': ADMIN_PREFIX + '/user_groups/{user_group_id}/members',
         'user_groups_new': ADMIN_PREFIX + '/user_groups/new',
         'user_groups_create': ADMIN_PREFIX + '/user_groups/create',
         'edit_user_group': ADMIN_PREFIX + '/user_groups/{user_group_id}/edit',
     }[name].format(**kwargs)
 
     if params:
         base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
     return base_url
 
 
 class TestAdminUserGroupsView(TestController):
 
     def test_show_users(self):
         self.log_user()
         self.app.get(route_path('user_groups'))
 
     def test_show_user_groups_data(self, xhr_header):
         self.log_user()
         response = self.app.get(route_path(
             'user_groups_data'), extra_environ=xhr_header)
 
         all_user_groups = UserGroup.query().count()
         assert response.json['recordsTotal'] == all_user_groups
 
     def test_show_user_groups_data_filtered(self, xhr_header):
         self.log_user()
         response = self.app.get(route_path(
             'user_groups_data', params={'search[value]': 'empty_search'}),
             extra_environ=xhr_header)
 
         all_user_groups = UserGroup.query().count()
         assert response.json['recordsTotal'] == all_user_groups
         assert response.json['recordsFiltered'] == 0
 
     def test_usergroup_escape(self, user_util, xhr_header):
         self.log_user()
 
         xss_img = '<img src="/image1" onload="alert(\'Hello, World!\');">'
         user = user_util.create_user()
         user.name = xss_img
         user.lastname = xss_img
         Session().add(user)
         Session().commit()
 
         user_group = user_util.create_user_group()
 
         user_group.users_group_name = xss_img
         user_group.user_group_description = '<strong onload="alert();">DESC</strong>'
 
         response = self.app.get(
             route_path('user_groups_data'), extra_environ=xhr_header)
 
         response.mustcontain(
             '&lt;strong onload=&#34;alert();&#34;&gt;DESC&lt;/strong&gt;')
         response.mustcontain(
             '&lt;img src=&#34;/image1&#34; onload=&#34;'
             'alert(&#39;Hello, World!&#39;);&#34;&gt;')
 
     def test_edit_user_group_autocomplete_empty_members(self, xhr_header, user_util):
         self.log_user()
         ug = user_util.create_user_group()
         response = self.app.get(
             route_path('user_group_members_data', user_group_id=ug.users_group_id),
             extra_environ=xhr_header)
 
         assert response.json == {'members': []}
 
     def test_edit_user_group_autocomplete_members(self, xhr_header, user_util):
         self.log_user()
         members = [u.user_id for u in User.get_all()]
         ug = user_util.create_user_group(members=members)
         response = self.app.get(
             route_path('user_group_members_data',
                        user_group_id=ug.users_group_id),
             extra_environ=xhr_header)
 
         assert len(response.json['members']) == len(members)
 
     def test_creation_page(self):
         self.log_user()
         self.app.get(route_path('user_groups_new'), status=200)
 
     def test_create(self):
         from rhodecode.lib import helpers as h
 
         self.log_user()
         users_group_name = 'test_user_group'
         response = self.app.post(route_path('user_groups_create'), {
             'users_group_name': users_group_name,
             'user_group_description': 'DESC',
             'active': True,
             'csrf_token': self.csrf_token})
 
         user_group_id = UserGroup.get_by_group_name(
             users_group_name).users_group_id
 
         user_group_link = h.link_to(
             users_group_name,
             route_path('edit_user_group', user_group_id=user_group_id))
 
         assert_session_flash(
             response,
             'Created user group %s' % user_group_link)
 
         fixture.destroy_user_group(users_group_name)
 
     def test_create_with_empty_name(self):
         self.log_user()
 
         response = self.app.post(route_path('user_groups_create'), {
             'users_group_name': '',
             'user_group_description': 'DESC',
             'active': True,
             'csrf_token': self.csrf_token}, status=200)
 
         response.mustcontain('Please enter a value')
 
     def test_create_duplicate(self, user_util):
         self.log_user()
 
         user_group = user_util.create_user_group()
         duplicate_name = user_group.users_group_name
         response = self.app.post(route_path('user_groups_create'), {
             'users_group_name': duplicate_name,
             'user_group_description': 'DESC',
             'active': True,
             'csrf_token': self.csrf_token}, status=200)
 
         response.mustcontain(
-            'User group `{}` already exists'.format(user_group.users_group_name))
+            'User group `{}` already exists'.format(duplicate_name))

rhodecode/apps/admin/tests/test_admin_users.py

@@ -1,781 +1,783 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2018 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import pytest
 from sqlalchemy.orm.exc import NoResultFound
 
 from rhodecode.lib import auth
 from rhodecode.lib import helpers as h
 from rhodecode.model.db import User, UserApiKeys, UserEmailMap, Repository
 from rhodecode.model.meta import Session
 from rhodecode.model.user import UserModel
 
 from rhodecode.tests import (
     TestController, TEST_USER_REGULAR_LOGIN, assert_session_flash)
 from rhodecode.tests.fixture import Fixture
 
 fixture = Fixture()
 
 
 def route_path(name, params=None, **kwargs):
     import urllib
     from rhodecode.apps._base import ADMIN_PREFIX
 
     base_url = {
         'users':
             ADMIN_PREFIX + '/users',
         'users_data':
             ADMIN_PREFIX + '/users_data',
         'users_create':
             ADMIN_PREFIX + '/users/create',
         'users_new':
             ADMIN_PREFIX + '/users/new',
         'user_edit':
             ADMIN_PREFIX + '/users/{user_id}/edit',
         'user_edit_advanced':
             ADMIN_PREFIX + '/users/{user_id}/edit/advanced',
         'user_edit_global_perms':
             ADMIN_PREFIX + '/users/{user_id}/edit/global_permissions',
         'user_edit_global_perms_update':
             ADMIN_PREFIX + '/users/{user_id}/edit/global_permissions/update',
         'user_update':
             ADMIN_PREFIX + '/users/{user_id}/update',
         'user_delete':
             ADMIN_PREFIX + '/users/{user_id}/delete',
         'user_force_password_reset':
             ADMIN_PREFIX + '/users/{user_id}/password_reset',
         'user_create_personal_repo_group':
             ADMIN_PREFIX + '/users/{user_id}/create_repo_group',
 
         'edit_user_auth_tokens':
             ADMIN_PREFIX + '/users/{user_id}/edit/auth_tokens',
         'edit_user_auth_tokens_add':
             ADMIN_PREFIX + '/users/{user_id}/edit/auth_tokens/new',
         'edit_user_auth_tokens_delete':
             ADMIN_PREFIX + '/users/{user_id}/edit/auth_tokens/delete',
 
         'edit_user_emails':
             ADMIN_PREFIX + '/users/{user_id}/edit/emails',
         'edit_user_emails_add':
             ADMIN_PREFIX + '/users/{user_id}/edit/emails/new',
         'edit_user_emails_delete':
             ADMIN_PREFIX + '/users/{user_id}/edit/emails/delete',
 
         'edit_user_ips':
             ADMIN_PREFIX + '/users/{user_id}/edit/ips',
         'edit_user_ips_add':
             ADMIN_PREFIX + '/users/{user_id}/edit/ips/new',
         'edit_user_ips_delete':
             ADMIN_PREFIX + '/users/{user_id}/edit/ips/delete',
 
         'edit_user_perms_summary':
             ADMIN_PREFIX + '/users/{user_id}/edit/permissions_summary',
         'edit_user_perms_summary_json':
             ADMIN_PREFIX + '/users/{user_id}/edit/permissions_summary/json',
 
         'edit_user_audit_logs':
             ADMIN_PREFIX + '/users/{user_id}/edit/audit',
 
     }[name].format(**kwargs)
 
     if params:
         base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
     return base_url
 
 
 class TestAdminUsersView(TestController):
 
     def test_show_users(self):
         self.log_user()
         self.app.get(route_path('users'))
 
     def test_show_users_data(self, xhr_header):
         self.log_user()
         response = self.app.get(route_path(
             'users_data'), extra_environ=xhr_header)
 
         all_users = User.query().filter(
             User.username != User.DEFAULT_USER).count()
         assert response.json['recordsTotal'] == all_users
 
     def test_show_users_data_filtered(self, xhr_header):
         self.log_user()
         response = self.app.get(route_path(
             'users_data', params={'search[value]': 'empty_search'}),
             extra_environ=xhr_header)
 
         all_users = User.query().filter(
             User.username != User.DEFAULT_USER).count()
         assert response.json['recordsTotal'] == all_users
         assert response.json['recordsFiltered'] == 0
 
     def test_auth_tokens_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.get(
             route_path('edit_user_auth_tokens', user_id=user.user_id),
             status=302)
 
     def test_auth_tokens(self):
         self.log_user()
 
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
+        user_id = user.user_id
+        auth_tokens = user.auth_tokens
         response = self.app.get(
-            route_path('edit_user_auth_tokens', user_id=user.user_id))
-        for token in user.auth_tokens:
+            route_path('edit_user_auth_tokens', user_id=user_id))
+        for token in auth_tokens:
             response.mustcontain(token)
             response.mustcontain('never')
 
     @pytest.mark.parametrize("desc, lifetime", [
         ('forever', -1),
         ('5mins', 60*5),
         ('30days', 60*60*24*30),
     ])
     def test_add_auth_token(self, desc, lifetime, user_util):
         self.log_user()
         user = user_util.create_user()
         user_id = user.user_id
 
         response = self.app.post(
             route_path('edit_user_auth_tokens_add', user_id=user_id),
             {'description': desc, 'lifetime': lifetime,
              'csrf_token': self.csrf_token})
         assert_session_flash(response, 'Auth token successfully created')
 
         response = response.follow()
         user = User.get(user_id)
         for auth_token in user.auth_tokens:
             response.mustcontain(auth_token)
 
     def test_delete_auth_token(self, user_util):
         self.log_user()
         user = user_util.create_user()
         user_id = user.user_id
         keys = user.auth_tokens
         assert 2 == len(keys)
 
         response = self.app.post(
             route_path('edit_user_auth_tokens_add', user_id=user_id),
             {'description': 'desc', 'lifetime': -1,
              'csrf_token': self.csrf_token})
         assert_session_flash(response, 'Auth token successfully created')
         response.follow()
 
         # now delete our key
         keys = UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all()
         assert 3 == len(keys)
 
         response = self.app.post(
             route_path('edit_user_auth_tokens_delete', user_id=user_id),
             {'del_auth_token': keys[0].user_api_key_id,
              'csrf_token': self.csrf_token})
 
         assert_session_flash(response, 'Auth token successfully deleted')
         keys = UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all()
         assert 2 == len(keys)
 
     def test_ips(self):
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         response = self.app.get(route_path('edit_user_ips', user_id=user.user_id))
         response.mustcontain('All IP addresses are allowed')
 
     @pytest.mark.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_ips_add(self, user_util, test_name, ip, ip_range, failure):
         self.log_user()
         user = user_util.create_user(username=test_name)
         user_id = user.user_id
 
         response = self.app.post(
             route_path('edit_user_ips_add', user_id=user_id),
             params={'new_ip': ip, 'csrf_token': self.csrf_token})
 
         if failure:
             assert_session_flash(
                 response, 'Please enter a valid IPv4 or IpV6 address')
             response = self.app.get(route_path('edit_user_ips', user_id=user_id))
 
             response.mustcontain(no=[ip])
             response.mustcontain(no=[ip_range])
 
         else:
             response = self.app.get(route_path('edit_user_ips', user_id=user_id))
             response.mustcontain(ip)
             response.mustcontain(ip_range)
 
     def test_ips_delete(self, user_util):
         self.log_user()
         user = user_util.create_user()
         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(route_path('edit_user_ips', user_id=user_id))
         response.mustcontain(ip)
         response.mustcontain(ip_range)
 
         self.app.post(
             route_path('edit_user_ips_delete', user_id=user_id),
             params={'del_ip_id': new_ip_id, 'csrf_token': self.csrf_token})
 
         response = self.app.get(route_path('edit_user_ips', user_id=user_id))
         response.mustcontain('All IP addresses are allowed')
         response.mustcontain(no=[ip])
         response.mustcontain(no=[ip_range])
 
     def test_emails(self):
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         response = self.app.get(
             route_path('edit_user_emails', user_id=user.user_id))
         response.mustcontain('No additional emails specified')
 
     def test_emails_add(self, user_util):
         self.log_user()
         user = user_util.create_user()
         user_id = user.user_id
 
         self.app.post(
             route_path('edit_user_emails_add', user_id=user_id),
             params={'new_email': '[e-mail]',
                     'csrf_token': self.csrf_token})
 
         response = self.app.get(
             route_path('edit_user_emails', user_id=user_id))
         response.mustcontain('[e-mail]')
 
     def test_emails_add_existing_email(self, user_util, user_regular):
         existing_email = user_regular.email
 
         self.log_user()
         user = user_util.create_user()
         user_id = user.user_id
 
         response = self.app.post(
             route_path('edit_user_emails_add', user_id=user_id),
             params={'new_email': existing_email,
                     'csrf_token': self.csrf_token})
         assert_session_flash(
             response, 'This e-mail address is already taken')
 
         response = self.app.get(
             route_path('edit_user_emails', user_id=user_id))
         response.mustcontain(no=[existing_email])
 
     def test_emails_delete(self, user_util):
         self.log_user()
         user = user_util.create_user()
         user_id = user.user_id
 
         self.app.post(
             route_path('edit_user_emails_add', user_id=user_id),
             params={'new_email': '[e-mail]',
                     'csrf_token': self.csrf_token})
 
         response = self.app.get(
             route_path('edit_user_emails', user_id=user_id))
         response.mustcontain('[e-mail]')
 
         user_email = UserEmailMap.query()\
             .filter(UserEmailMap.email == '[e-mail]') \
             .filter(UserEmailMap.user_id == user_id)\
             .one()
 
         del_email_id = user_email.email_id
         self.app.post(
             route_path('edit_user_emails_delete', user_id=user_id),
             params={'del_email_id': del_email_id,
                     'csrf_token': self.csrf_token})
 
         response = self.app.get(
             route_path('edit_user_emails', user_id=user_id))
         response.mustcontain(no=['[e-mail]'])
 
 
     def test_create(self, request, xhr_header):
         self.log_user()
         username = 'newtestuser'
         password = 'test12'
         password_confirmation = password
         name = 'name'
         lastname = 'lastname'
         email = '[e-mail]'
 
         self.app.get(route_path('users_new'))
 
         response = self.app.post(route_path('users_create'), params={
             'username': username,
             'password': password,
             'password_confirmation': password_confirmation,
             'firstname': name,
             'active': True,
             'lastname': lastname,
             'extern_name': 'rhodecode',
             'extern_type': 'rhodecode',
             'email': email,
             'csrf_token': self.csrf_token,
         })
         user_link = h.link_to(
             username,
             route_path(
                 'user_edit', user_id=User.get_by_username(username).user_id))
         assert_session_flash(response, 'Created user %s' % (user_link,))
 
         @request.addfinalizer
         def cleanup():
             fixture.destroy_user(username)
             Session().commit()
 
         new_user = User.query().filter(User.username == username).one()
 
         assert new_user.username == username
         assert auth.check_password(password, new_user.password)
         assert new_user.name == name
         assert new_user.lastname == lastname
         assert new_user.email == email
 
         response = self.app.get(route_path('users_data'),
                                 extra_environ=xhr_header)
         response.mustcontain(username)
 
     def test_create_err(self):
         self.log_user()
         username = 'new_user'
         password = ''
         name = 'name'
         lastname = 'lastname'
         email = 'errmail.com'
 
         self.app.get(route_path('users_new'))
 
         response = self.app.post(route_path('users_create'), params={
             'username': username,
             'password': password,
             'name': name,
             'active': False,
             'lastname': lastname,
             'email': email,
             'csrf_token': self.csrf_token,
         })
 
         msg = u'Username "%(username)s" is forbidden'
         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()
 
     def test_new(self):
         self.log_user()
         self.app.get(route_path('users_new'))
 
     @pytest.mark.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]'}),
         ('language', {'language': 'de'}),
         ('language', {'language': 'en'}),
         # ('new_password', {'new_password': 'foobar123',
         #                   'password_confirmation': 'foobar123'})
         ])
     def test_update(self, name, attrs, user_util):
         self.log_user()
         usr = user_util.create_user(
             password='qweqwe',
             email='[e-mail]',
             extern_type='rhodecode',
             extern_name='xxx',
         )
         user_id = usr.user_id
         Session().commit()
 
         params = usr.get_api_data()
         cur_lang = params['language'] or 'en'
         params.update({
             'password_confirmation': '',
             'new_password': '',
             'language': cur_lang,
             'csrf_token': self.csrf_token,
         })
         params.update({'new_password': ''})
         params.update(attrs)
         if name == 'email':
             params['emails'] = [attrs['email']]
         elif name == 'extern_type':
             # cannot update this via form, expected value is original one
             params['extern_type'] = "rhodecode"
         elif name == 'extern_name':
             # cannot update this via form, expected value is original one
             params['extern_name'] = 'xxx'
             # special case since this user is not
             # logged in yet his data is not filled
             # so we use creation data
 
         response = self.app.post(
             route_path('user_update', user_id=usr.user_id), params)
         assert response.status_int == 302
         assert_session_flash(response, 'User updated successfully')
 
         updated_user = User.get(user_id)
         updated_params = updated_user.get_api_data()
         updated_params.update({'password_confirmation': ''})
         updated_params.update({'new_password': ''})
 
         del params['csrf_token']
         assert params == updated_params
 
     def test_update_and_migrate_password(
             self, autologin_user, real_crypto_backend, user_util):
 
         user = user_util.create_user()
         temp_user = user.username
         user.password = auth._RhodeCodeCryptoSha256().hash_create(
             b'test123')
         Session().add(user)
         Session().commit()
 
         params = user.get_api_data()
 
         params.update({
             'password_confirmation': 'qweqwe123',
             'new_password': 'qweqwe123',
             'language': 'en',
             'csrf_token': autologin_user.csrf_token,
         })
 
         response = self.app.post(
             route_path('user_update', user_id=user.user_id), params)
         assert response.status_int == 302
         assert_session_flash(response, 'User updated successfully')
 
         # new password should be bcrypted, after log-in and transfer
         user = User.get_by_username(temp_user)
         assert user.password.startswith('$')
 
         updated_user = User.get_by_username(temp_user)
         updated_params = updated_user.get_api_data()
         updated_params.update({'password_confirmation': 'qweqwe123'})
         updated_params.update({'new_password': 'qweqwe123'})
 
         del params['csrf_token']
         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(
             route_path('user_delete', user_id=new_user.user_id),
             params={'csrf_token': self.csrf_token})
 
         assert_session_flash(response, 'Successfully deleted user')
 
     def test_delete_owner_of_repository(self, request, user_util):
         self.log_user()
         obj_name = 'test_repo'
         usr = user_util.create_user()
         username = usr.username
         fixture.create_repo(obj_name, cur_user=usr.username)
 
         new_user = Session().query(User)\
             .filter(User.username == username).one()
         response = self.app.post(
             route_path('user_delete', user_id=new_user.user_id),
             params={'csrf_token': self.csrf_token})
 
         msg = 'user "%s" still owns 1 repositories and cannot be removed. ' \
               'Switch owners or remove those repositories:%s' % (username,
                                                                  obj_name)
         assert_session_flash(response, msg)
         fixture.destroy_repo(obj_name)
 
     def test_delete_owner_of_repository_detaching(self, request, user_util):
         self.log_user()
         obj_name = 'test_repo'
         usr = user_util.create_user(auto_cleanup=False)
         username = usr.username
         fixture.create_repo(obj_name, cur_user=usr.username)
 
         new_user = Session().query(User)\
             .filter(User.username == username).one()
         response = self.app.post(
             route_path('user_delete', user_id=new_user.user_id),
             params={'user_repos': 'detach', 'csrf_token': self.csrf_token})
 
         msg = 'Detached 1 repositories'
         assert_session_flash(response, msg)
         fixture.destroy_repo(obj_name)
 
     def test_delete_owner_of_repository_deleting(self, request, user_util):
         self.log_user()
         obj_name = 'test_repo'
         usr = user_util.create_user(auto_cleanup=False)
         username = usr.username
         fixture.create_repo(obj_name, cur_user=usr.username)
 
         new_user = Session().query(User)\
             .filter(User.username == username).one()
         response = self.app.post(
             route_path('user_delete', user_id=new_user.user_id),
             params={'user_repos': 'delete', 'csrf_token': self.csrf_token})
 
         msg = 'Deleted 1 repositories'
         assert_session_flash(response, msg)
 
     def test_delete_owner_of_repository_group(self, request, user_util):
         self.log_user()
         obj_name = 'test_group'
         usr = user_util.create_user()
         username = usr.username
         fixture.create_repo_group(obj_name, cur_user=usr.username)
 
         new_user = Session().query(User)\
             .filter(User.username == username).one()
         response = self.app.post(
             route_path('user_delete', user_id=new_user.user_id),
             params={'csrf_token': self.csrf_token})
 
         msg = 'user "%s" still owns 1 repository groups and cannot be removed. ' \
               'Switch owners or remove those repository groups:%s' % (username,
                                                                       obj_name)
         assert_session_flash(response, msg)
         fixture.destroy_repo_group(obj_name)
 
     def test_delete_owner_of_repository_group_detaching(self, request, user_util):
         self.log_user()
         obj_name = 'test_group'
         usr = user_util.create_user(auto_cleanup=False)
         username = usr.username
         fixture.create_repo_group(obj_name, cur_user=usr.username)
 
         new_user = Session().query(User)\
             .filter(User.username == username).one()
         response = self.app.post(
             route_path('user_delete', user_id=new_user.user_id),
             params={'user_repo_groups': 'delete', 'csrf_token': self.csrf_token})
 
         msg = 'Deleted 1 repository groups'
         assert_session_flash(response, msg)
 
     def test_delete_owner_of_repository_group_deleting(self, request, user_util):
         self.log_user()
         obj_name = 'test_group'
         usr = user_util.create_user(auto_cleanup=False)
         username = usr.username
         fixture.create_repo_group(obj_name, cur_user=usr.username)
 
         new_user = Session().query(User)\
             .filter(User.username == username).one()
         response = self.app.post(
             route_path('user_delete', user_id=new_user.user_id),
             params={'user_repo_groups': 'detach', 'csrf_token': self.csrf_token})
 
         msg = 'Detached 1 repository groups'
         assert_session_flash(response, msg)
         fixture.destroy_repo_group(obj_name)
 
     def test_delete_owner_of_user_group(self, request, user_util):
         self.log_user()
         obj_name = 'test_user_group'
         usr = user_util.create_user()
         username = usr.username
         fixture.create_user_group(obj_name, cur_user=usr.username)
 
         new_user = Session().query(User)\
             .filter(User.username == username).one()
         response = self.app.post(
             route_path('user_delete', user_id=new_user.user_id),
             params={'csrf_token': self.csrf_token})
 
         msg = 'user "%s" still owns 1 user groups and cannot be removed. ' \
               'Switch owners or remove those user groups:%s' % (username,
                                                                 obj_name)
         assert_session_flash(response, msg)
         fixture.destroy_user_group(obj_name)
 
     def test_delete_owner_of_user_group_detaching(self, request, user_util):
         self.log_user()
         obj_name = 'test_user_group'
         usr = user_util.create_user(auto_cleanup=False)
         username = usr.username
         fixture.create_user_group(obj_name, cur_user=usr.username)
 
         new_user = Session().query(User)\
             .filter(User.username == username).one()
         try:
             response = self.app.post(
                 route_path('user_delete', user_id=new_user.user_id),
                 params={'user_user_groups': 'detach',
                         'csrf_token': self.csrf_token})
 
             msg = 'Detached 1 user groups'
             assert_session_flash(response, msg)
         finally:
             fixture.destroy_user_group(obj_name)
 
     def test_delete_owner_of_user_group_deleting(self, request, user_util):
         self.log_user()
         obj_name = 'test_user_group'
         usr = user_util.create_user(auto_cleanup=False)
         username = usr.username
         fixture.create_user_group(obj_name, cur_user=usr.username)
 
         new_user = Session().query(User)\
             .filter(User.username == username).one()
         response = self.app.post(
             route_path('user_delete', user_id=new_user.user_id),
             params={'user_user_groups': 'delete', 'csrf_token': self.csrf_token})
 
         msg = 'Deleted 1 user groups'
         assert_session_flash(response, msg)
 
     def test_edit(self, user_util):
         self.log_user()
         user = user_util.create_user()
         self.app.get(route_path('user_edit', user_id=user.user_id))
 
     def test_edit_default_user_redirect(self):
         self.log_user()
         user = User.get_default_user()
         self.app.get(route_path('user_edit', user_id=user.user_id), status=302)
 
     @pytest.mark.parametrize(
         'repo_create, repo_create_write, user_group_create, repo_group_create,'
         'fork_create, inherit_default_permissions, expect_error,'
         'expect_form_error', [
             ('hg.create.none', 'hg.create.write_on_repogroup.false',
              'hg.usergroup.create.false', 'hg.repogroup.create.false',
              'hg.fork.none', 'hg.inherit_default_perms.false', False, False),
             ('hg.create.repository', 'hg.create.write_on_repogroup.false',
              'hg.usergroup.create.false', 'hg.repogroup.create.false',
              'hg.fork.none', 'hg.inherit_default_perms.false', False, False),
             ('hg.create.repository', 'hg.create.write_on_repogroup.true',
              'hg.usergroup.create.true', 'hg.repogroup.create.true',
              'hg.fork.repository', 'hg.inherit_default_perms.false', False,
              False),
             ('hg.create.XXX', 'hg.create.write_on_repogroup.true',
              'hg.usergroup.create.true', 'hg.repogroup.create.true',
              'hg.fork.repository', 'hg.inherit_default_perms.false', False,
              True),
             ('', '', '', '', '', '', True, False),
         ])
     def test_global_perms_on_user(
             self, repo_create, repo_create_write, user_group_create,
             repo_group_create, fork_create, expect_error, expect_form_error,
             inherit_default_permissions, user_util):
         self.log_user()
         user = user_util.create_user()
         uid = user.user_id
 
         # ENABLE REPO CREATE ON A GROUP
         perm_params = {
             'inherit_default_permissions': False,
             'default_repo_create': repo_create,
             'default_repo_create_on_write': repo_create_write,
             'default_user_group_create': user_group_create,
             'default_repo_group_create': repo_group_create,
             'default_fork_create': fork_create,
             'default_inherit_default_permissions': inherit_default_permissions,
             'csrf_token': self.csrf_token,
         }
         response = self.app.post(
             route_path('user_edit_global_perms_update', user_id=uid),
             params=perm_params)
 
         if expect_form_error:
             assert response.status_int == 200
             response.mustcontain('Value must be one of')
         else:
             if expect_error:
                 msg = 'An error occurred during permissions saving'
             else:
                 msg = 'User global permissions updated successfully'
                 ug = User.get(uid)
                 del perm_params['inherit_default_permissions']
                 del perm_params['csrf_token']
                 assert perm_params == ug.get_default_perms()
             assert_session_flash(response, msg)
 
     def test_global_permissions_initial_values(self, user_util):
         self.log_user()
         user = user_util.create_user()
         uid = user.user_id
         response = self.app.get(
             route_path('user_edit_global_perms', user_id=uid))
         default_user = User.get_default_user()
         default_permissions = default_user.get_default_perms()
         assert_response = response.assert_response()
         expected_permissions = (
             'default_repo_create', 'default_repo_create_on_write',
             'default_fork_create', 'default_repo_group_create',
             'default_user_group_create', 'default_inherit_default_permissions')
         for permission in expected_permissions:
             css_selector = '[name={}][checked=checked]'.format(permission)
             element = assert_response.get_element(css_selector)
             assert element.value == default_permissions[permission]
 
     def test_perms_summary_page(self):
         user = self.log_user()
         response = self.app.get(
             route_path('edit_user_perms_summary', user_id=user['user_id']))
         for repo in Repository.query().all():
             response.mustcontain(repo.repo_name)
 
     def test_perms_summary_page_json(self):
         user = self.log_user()
         response = self.app.get(
             route_path('edit_user_perms_summary_json', user_id=user['user_id']))
         for repo in Repository.query().all():
             response.mustcontain(repo.repo_name)
 
     def test_audit_log_page(self):
         user = self.log_user()
         self.app.get(
             route_path('edit_user_audit_logs', user_id=user['user_id']))