grids: fixed tests and added one for permissions to repository group.

rhodecode/apps/admin/tests/test_admin_repos.py

@@ -1,512 +1,514 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2019 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',
+        'repos_data': ADMIN_PREFIX + '/repos_data',
         '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):
+    def test_repo_list(self, autologin_user, user_util, xhr_header):
         repo = user_util.create_repo()
         repo_name = repo.repo_name
         response = self.app.get(
-            route_path('repos'), status=200)
+            route_path('repos_data'), status=200,
+            extra_environ=xhr_header)
 
         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 = response.assert_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 = response.assert_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=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=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=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_repository_groups.py

@@ -1,194 +1,194 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2019 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 os
 import pytest
 
 from rhodecode.apps._base import ADMIN_PREFIX
 from rhodecode.lib import helpers as h
 from rhodecode.model.db import Repository, UserRepoToPerm, User, RepoGroup
 from rhodecode.model.meta import Session
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.tests import (
     assert_session_flash, TEST_USER_REGULAR_LOGIN, TESTS_TMP_PATH)
 from rhodecode.tests.fixture import Fixture
 
 fixture = Fixture()
 
 
 def route_path(name, params=None, **kwargs):
     import urllib
 
     base_url = {
         'repo_groups': ADMIN_PREFIX + '/repo_groups',
         'repo_groups_data': ADMIN_PREFIX + '/repo_groups_data',
         'repo_group_new': ADMIN_PREFIX + '/repo_group/new',
         'repo_group_create': ADMIN_PREFIX + '/repo_group/create',
 
     }[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 TestAdminRepositoryGroups(object):
 
     def test_show_repo_groups(self, autologin_user):
         self.app.get(route_path('repo_groups'))
 
     def test_show_repo_groups_data(self, autologin_user, xhr_header):
         response = self.app.get(route_path(
             'repo_groups_data'), extra_environ=xhr_header)
 
         all_repo_groups = RepoGroup.query().count()
         assert response.json['recordsTotal'] == all_repo_groups
 
     def test_show_repo_groups_data_filtered(self, autologin_user, xhr_header):
         response = self.app.get(route_path(
             'repo_groups_data', params={'search[value]': 'empty_search'}),
             extra_environ=xhr_header)
 
         all_repo_groups = RepoGroup.query().count()
         assert response.json['recordsTotal'] == all_repo_groups
         assert response.json['recordsFiltered'] == 0
 
     def test_show_repo_groups_after_creating_group(self, autologin_user, xhr_header):
         fixture.create_repo_group('test_repo_group')
         response = self.app.get(route_path(
             'repo_groups_data'), extra_environ=xhr_header)
-        response.mustcontain('"name_raw": "test_repo_group"')
+        response.mustcontain('<a href=\\"/{}/_edit\\" title=\\"Edit\\">Edit</a>'.format('test_repo_group'))
         fixture.destroy_repo_group('test_repo_group')
 
     def test_new(self, autologin_user):
         self.app.get(route_path('repo_group_new'))
 
     def test_new_with_parent_group(self, autologin_user, user_util):
         gr = user_util.create_repo_group()
 
         self.app.get(route_path('repo_group_new'),
                      params=dict(parent_group=gr.group_name))
 
     def test_new_by_regular_user_no_permission(self, autologin_regular_user):
         self.app.get(route_path('repo_group_new'), status=403)
 
     @pytest.mark.parametrize('repo_group_name', [
         'git_repo',
         'git_repo_ąć',
         'hg_repo',
         '12345',
         'hg_repo_ąć',
     ])
     def test_create(self, autologin_user, repo_group_name, csrf_token):
         repo_group_name_unicode = repo_group_name.decode('utf8')
         description = 'description for newly created repo group'
 
         response = self.app.post(
             route_path('repo_group_create'),
             fixture._get_group_create_params(
                 group_name=repo_group_name,
                 group_description=description,
                 csrf_token=csrf_token))
 
         # run the check page that triggers the flash message
         repo_gr_url = h.route_path(
             'repo_group_home', repo_group_name=repo_group_name)
 
         assert_session_flash(
             response,
             'Created repository group <a href="%s">%s</a>' % (
                 repo_gr_url, repo_group_name_unicode))
 
         # # test if the repo group was created in the database
         new_repo_group = RepoGroupModel()._get_repo_group(
             repo_group_name_unicode)
         assert new_repo_group is not None
 
         assert new_repo_group.group_name == repo_group_name_unicode
         assert new_repo_group.group_description == description
 
         # test if the repository is visible in the list ?
         response = self.app.get(repo_gr_url)
         response.mustcontain(repo_group_name)
 
         # test if the repository group was created on filesystem
         is_on_filesystem = os.path.isdir(
             os.path.join(TESTS_TMP_PATH, repo_group_name))
         if not is_on_filesystem:
             self.fail('no repo group %s in filesystem' % repo_group_name)
 
         RepoGroupModel().delete(repo_group_name_unicode)
         Session().commit()
 
     @pytest.mark.parametrize('repo_group_name', [
         'git_repo',
         'git_repo_ąć',
         'hg_repo',
         '12345',
         'hg_repo_ąć',
     ])
     def test_create_subgroup(self, autologin_user, user_util, repo_group_name, csrf_token):
         parent_group = user_util.create_repo_group()
         parent_group_name = parent_group.group_name
 
         expected_group_name = '{}/{}'.format(
             parent_group_name, repo_group_name)
         expected_group_name_unicode = expected_group_name.decode('utf8')
 
         try:
             response = self.app.post(
                 route_path('repo_group_create'),
                 fixture._get_group_create_params(
                     group_name=repo_group_name,
                     group_parent_id=parent_group.group_id,
                     group_description='Test desciption',
                     csrf_token=csrf_token))
 
             assert_session_flash(
                 response,
                 u'Created repository group <a href="%s">%s</a>' % (
                     h.route_path('repo_group_home',
                                  repo_group_name=expected_group_name),
                     expected_group_name_unicode))
         finally:
             RepoGroupModel().delete(expected_group_name_unicode)
             Session().commit()
 
     def test_user_with_creation_permissions_cannot_create_subgroups(
             self, autologin_regular_user, user_util):
 
         user_util.grant_user_permission(
             TEST_USER_REGULAR_LOGIN, 'hg.repogroup.create.true')
         parent_group = user_util.create_repo_group()
         parent_group_id = parent_group.group_id
         self.app.get(
             route_path('repo_group_new',
                        params=dict(parent_group=parent_group_id), ),
             status=403)

rhodecode/apps/admin/tests/test_admin_users.py

@@ -1,793 +1,794 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2019 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_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',
 
         'edit_user_audit_logs_download':
             ADMIN_PREFIX + '/users/{user_id}/edit/audit/download',
 
     }[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_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,
             'description': 'mr CTO',
             '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,
             'description': 'mr CTO',
             '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'}),
         ('description', {'description': 'hello CTO'}),
         # ('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 `{}`'.format(username))
 
     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)
+        Session().commit()
 
         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']))
 
     def test_audit_log_page_download(self):
         user = self.log_user()
         user_id = user['user_id']
         response = self.app.get(
             route_path('edit_user_audit_logs_download', user_id=user_id))
 
         assert response.content_disposition == \
                'attachment; filename=user_{}_audit_logs.json'.format(user_id)
         assert response.content_type == "application/json"

rhodecode/apps/home/tests/test_home.py

@@ -1,127 +1,179 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2019 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
 
 import rhodecode
-from rhodecode.model.db import Repository
+from rhodecode.model.db import Repository, RepoGroup, User
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.settings import SettingsModel
 from rhodecode.tests import TestController
 from rhodecode.tests.fixture import Fixture
 from rhodecode.lib import helpers as h
 
 fixture = Fixture()
 
 
 def route_path(name, **kwargs):
     return {
         'home': '/',
+        'main_page_repos_data': '/_home_repos',
+        'main_page_repo_groups_data': '/_home_repo_groups',
         'repo_group_home': '/{repo_group_name}'
     }[name].format(**kwargs)
 
 
 class TestHomeController(TestController):
 
     def test_index(self):
         self.log_user()
         response = self.app.get(route_path('home'))
         # if global permission is set
         response.mustcontain('New Repository')
 
+    def test_index_grid_repos(self, xhr_header):
+        self.log_user()
+        response = self.app.get(route_path('main_page_repos_data'), extra_environ=xhr_header)
         # search for objects inside the JavaScript JSON
-        for repo in Repository.getAll():
-            response.mustcontain('"name_raw": "%s"' % repo.repo_name)
+        for obj in Repository.getAll():
+            response.mustcontain('<a href=\\"/{}\\">'.format(obj.repo_name))
+
+    def test_index_grid_repo_groups(self, xhr_header):
+        self.log_user()
+        response = self.app.get(route_path('main_page_repo_groups_data'),
+                                extra_environ=xhr_header,)
+
+        # search for objects inside the JavaScript JSON
+        for obj in RepoGroup.getAll():
+            response.mustcontain('<a href=\\"/{}\\">'.format(obj.group_name))
+
+    def test_index_grid_repo_groups_without_access(self, xhr_header, user_util):
+        user = user_util.create_user(password='qweqwe')
+        group_ok = user_util.create_repo_group(owner=user)
+        group_id_ok = group_ok.group_id
+
+        group_forbidden = user_util.create_repo_group(owner=User.get_first_super_admin())
+        group_id_forbidden = group_forbidden.group_id
+
+        user_util.grant_user_permission_to_repo_group(group_forbidden, user, 'group.none')
+        self.log_user(user.username, 'qweqwe')
+
+        self.app.get(route_path('main_page_repo_groups_data'),
+                     extra_environ=xhr_header,
+                     params={'repo_group_id': group_id_ok}, status=200)
+
+        self.app.get(route_path('main_page_repo_groups_data'),
+                     extra_environ=xhr_header,
+                     params={'repo_group_id': group_id_forbidden}, status=404)
 
     def test_index_contains_statics_with_ver(self):
         from rhodecode.lib.base import calculate_version_hash
 
         self.log_user()
         response = self.app.get(route_path('home'))
 
         rhodecode_version_hash = calculate_version_hash(
             {'beaker.session.secret': 'test-rc-uytcxaz'})
         response.mustcontain('style.css?ver={0}'.format(rhodecode_version_hash))
         response.mustcontain('scripts.min.js?ver={0}'.format(rhodecode_version_hash))
 
-    def test_index_contains_backend_specific_details(self, backend):
+    def test_index_contains_backend_specific_details(self, backend, xhr_header):
         self.log_user()
-        response = self.app.get(route_path('home'))
+        response = self.app.get(route_path('main_page_repos_data'), extra_environ=xhr_header)
         tip = backend.repo.get_commit().raw_id
 
         # html in javascript variable:
         response.mustcontain(r'<i class=\"icon-%s\"' % (backend.alias, ))
         response.mustcontain(r'href=\"/%s\"' % (backend.repo_name, ))
 
         response.mustcontain("""/%s/changeset/%s""" % (backend.repo_name, tip))
         response.mustcontain("""Added a symlink""")
 
     def test_index_with_anonymous_access_disabled(self):
         with fixture.anon_access(False):
             response = self.app.get(route_path('home'), status=302)
             assert 'login' in response.location
 
-    def test_index_page_on_groups(self, autologin_user, repo_group):
-        response = self.app.get(route_path('repo_group_home', repo_group_name='gr1'))
-        response.mustcontain("gr1/repo_in_group")
+    def test_index_page_on_groups_with_wrong_group_id(self, autologin_user, xhr_header):
+        group_id = 918123
+        self.app.get(
+            route_path('main_page_repo_groups_data'),
+            params={'repo_group_id': group_id},
+            status=404, extra_environ=xhr_header)
 
-    def test_index_page_on_group_with_trailing_slash(
-            self, autologin_user, repo_group):
-        response = self.app.get(route_path('repo_group_home', repo_group_name='gr1') + '/')
-        response.mustcontain("gr1/repo_in_group")
+    def test_index_page_on_groups(self, autologin_user, user_util, xhr_header):
+        gr = user_util.create_repo_group()
+        repo = user_util.create_repo(parent=gr)
+        repo_name = repo.repo_name
+        group_id = gr.group_id
+
+        response = self.app.get(route_path(
+            'repo_group_home', repo_group_name=gr.group_name))
+        response.mustcontain('d.repo_group_id = {}'.format(group_id))
 
-    @pytest.fixture(scope='class')
-    def repo_group(self, request):
-        gr = fixture.create_repo_group('gr1')
-        fixture.create_repo(name='gr1/repo_in_group', repo_group=gr)
+        response = self.app.get(
+            route_path('main_page_repos_data'),
+            params={'repo_group_id': group_id},
+            extra_environ=xhr_header,)
+        response.mustcontain(repo_name)
 
-        @request.addfinalizer
-        def cleanup():
-            RepoModel().delete('gr1/repo_in_group')
-            RepoGroupModel().delete(repo_group='gr1', force_delete=True)
-            Session().commit()
+    def test_index_page_on_group_with_trailing_slash(self, autologin_user, user_util, xhr_header):
+        gr = user_util.create_repo_group()
+        repo = user_util.create_repo(parent=gr)
+        repo_name = repo.repo_name
+        group_id = gr.group_id
+
+        response = self.app.get(route_path(
+            'repo_group_home', repo_group_name=gr.group_name+'/'))
+        response.mustcontain('d.repo_group_id = {}'.format(group_id))
+
+        response = self.app.get(
+            route_path('main_page_repos_data'),
+            params={'repo_group_id': group_id},
+            extra_environ=xhr_header, )
+        response.mustcontain(repo_name)
 
     @pytest.mark.parametrize("name, state", [
         ('Disabled', False),
         ('Enabled', True),
     ])
     def test_index_show_version(self, autologin_user, name, state):
         version_string = 'RhodeCode Enterprise %s' % rhodecode.__version__
 
         sett = SettingsModel().create_or_update_setting(
             'show_version', state, 'bool')
         Session().add(sett)
         Session().commit()
         SettingsModel().invalidate_settings_cache()
 
         response = self.app.get(route_path('home'))
         if state is True:
             response.mustcontain(version_string)
         if state is False:
             response.mustcontain(no=[version_string])
 
     def test_logout_form_contains_csrf(self, autologin_user, csrf_token):
         response = self.app.get(route_path('home'))
         assert_response = response.assert_response()
         element = assert_response.get_element('.logout [name=csrf_token]')
         assert element.value == csrf_token

rhodecode/apps/home/views.py

@@ -1,823 +1,822 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2019 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 re
 import logging
 import collections
 
 from pyramid.httpexceptions import HTTPNotFound
 from pyramid.view import view_config
 
 from rhodecode.apps._base import BaseAppView, DataGridAppView
 from rhodecode.lib import helpers as h
 from rhodecode.lib.auth import (
     LoginRequired, NotAnonymous, HasRepoGroupPermissionAnyDecorator, CSRFRequired,
-    HasRepoGroupPermissionAny)
+    HasRepoGroupPermissionAny, AuthUser)
 from rhodecode.lib.codeblocks import filenode_as_lines_tokens
 from rhodecode.lib.index import searcher_from_config
 from rhodecode.lib.utils2 import safe_unicode, str2bool, safe_int
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.model.db import (
     func, true, or_, case, in_filter_generator, Session,
     Repository, RepoGroup, User, UserGroup)
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.user import UserModel
 from rhodecode.model.user_group import UserGroupModel
 
 log = logging.getLogger(__name__)
 
 
 class HomeView(BaseAppView, DataGridAppView):
 
     def load_default_context(self):
         c = self._get_local_tmpl_context()
         c.user = c.auth_user.get_instance()
 
         return c
 
     @LoginRequired()
     @view_config(
         route_name='user_autocomplete_data', request_method='GET',
         renderer='json_ext', xhr=True)
     def user_autocomplete_data(self):
         self.load_default_context()
         query = self.request.GET.get('query')
         active = str2bool(self.request.GET.get('active') or True)
         include_groups = str2bool(self.request.GET.get('user_groups'))
         expand_groups = str2bool(self.request.GET.get('user_groups_expand'))
         skip_default_user = str2bool(self.request.GET.get('skip_default_user'))
 
         log.debug('generating user list, query:%s, active:%s, with_groups:%s',
                   query, active, include_groups)
 
         _users = UserModel().get_users(
             name_contains=query, only_active=active)
 
         def maybe_skip_default_user(usr):
             if skip_default_user and usr['username'] == UserModel.cls.DEFAULT_USER:
                 return False
             return True
         _users = filter(maybe_skip_default_user, _users)
 
         if include_groups:
             # extend with user groups
             _user_groups = UserGroupModel().get_user_groups(
                 name_contains=query, only_active=active,
                 expand_groups=expand_groups)
             _users = _users + _user_groups
 
         return {'suggestions': _users}
 
     @LoginRequired()
     @NotAnonymous()
     @view_config(
         route_name='user_group_autocomplete_data', request_method='GET',
         renderer='json_ext', xhr=True)
     def user_group_autocomplete_data(self):
         self.load_default_context()
         query = self.request.GET.get('query')
         active = str2bool(self.request.GET.get('active') or True)
         expand_groups = str2bool(self.request.GET.get('user_groups_expand'))
 
         log.debug('generating user group list, query:%s, active:%s',
                   query, active)
 
         _user_groups = UserGroupModel().get_user_groups(
             name_contains=query, only_active=active,
             expand_groups=expand_groups)
         _user_groups = _user_groups
 
         return {'suggestions': _user_groups}
 
     def _get_repo_list(self, name_contains=None, repo_type=None, repo_group_name='', limit=20):
         org_query = name_contains
         allowed_ids = self._rhodecode_user.repo_acl_ids(
             ['repository.read', 'repository.write', 'repository.admin'],
             cache=False, name_filter=name_contains) or [-1]
 
         query = Repository.query()\
             .filter(Repository.archived.isnot(true()))\
             .filter(or_(
                 # generate multiple IN to fix limitation problems
                 *in_filter_generator(Repository.repo_id, allowed_ids)
             ))
 
         query = query.order_by(case(
             [
                 (Repository.repo_name.startswith(repo_group_name), repo_group_name+'/'),
             ],
         ))
         query = query.order_by(func.length(Repository.repo_name))
         query = query.order_by(Repository.repo_name)
 
         if repo_type:
             query = query.filter(Repository.repo_type == repo_type)
 
         if name_contains:
             ilike_expression = u'%{}%'.format(safe_unicode(name_contains))
             query = query.filter(
                 Repository.repo_name.ilike(ilike_expression))
             query = query.limit(limit)
 
         acl_iter = query
 
         return [
             {
                 'id': obj.repo_name,
                 'value': org_query,
                 'value_display': obj.repo_name,
                 'text': obj.repo_name,
                 'type': 'repo',
                 'repo_id': obj.repo_id,
                 'repo_type': obj.repo_type,
                 'private': obj.private,
                 'url': h.route_path('repo_summary', repo_name=obj.repo_name)
             }
             for obj in acl_iter]
 
     def _get_repo_group_list(self, name_contains=None, repo_group_name='', limit=20):
         org_query = name_contains
         allowed_ids = self._rhodecode_user.repo_group_acl_ids(
             ['group.read', 'group.write', 'group.admin'],
             cache=False, name_filter=name_contains) or [-1]
 
         query = RepoGroup.query()\
             .filter(or_(
                 # generate multiple IN to fix limitation problems
                 *in_filter_generator(RepoGroup.group_id, allowed_ids)
             ))
 
         query = query.order_by(case(
             [
                 (RepoGroup.group_name.startswith(repo_group_name), repo_group_name+'/'),
             ],
         ))
         query = query.order_by(func.length(RepoGroup.group_name))
         query = query.order_by(RepoGroup.group_name)
 
         if name_contains:
             ilike_expression = u'%{}%'.format(safe_unicode(name_contains))
             query = query.filter(
                 RepoGroup.group_name.ilike(ilike_expression))
             query = query.limit(limit)
 
         acl_iter = query
 
         return [
             {
                 'id': obj.group_name,
                 'value': org_query,
                 'value_display': obj.group_name,
                 'text': obj.group_name,
                 'type': 'repo_group',
                 'repo_group_id': obj.group_id,
                 'url': h.route_path(
                     'repo_group_home', repo_group_name=obj.group_name)
             }
             for obj in acl_iter]
 
     def _get_user_list(self, name_contains=None, limit=20):
         org_query = name_contains
         if not name_contains:
             return [], False
 
         # TODO(marcink): should all logged in users be allowed to search others?
         allowed_user_search = self._rhodecode_user.username != User.DEFAULT_USER
         if not allowed_user_search:
             return [], False
 
         name_contains = re.compile('(?:user:[ ]?)(.+)').findall(name_contains)
         if len(name_contains) != 1:
             return [], False
 
         name_contains = name_contains[0]
 
         query = User.query()\
             .order_by(func.length(User.username))\
             .order_by(User.username) \
             .filter(User.username != User.DEFAULT_USER)
 
         if name_contains:
             ilike_expression = u'%{}%'.format(safe_unicode(name_contains))
             query = query.filter(
                 User.username.ilike(ilike_expression))
             query = query.limit(limit)
 
         acl_iter = query
 
         return [
             {
                 'id': obj.user_id,
                 'value': org_query,
                 'value_display': 'user: `{}`'.format(obj.username),
                 'type': 'user',
                 'icon_link': h.gravatar_url(obj.email, 30),
                 'url': h.route_path(
                     'user_profile', username=obj.username)
             }
             for obj in acl_iter], True
 
     def _get_user_groups_list(self, name_contains=None, limit=20):
         org_query = name_contains
         if not name_contains:
             return [], False
 
         # TODO(marcink): should all logged in users be allowed to search others?
         allowed_user_search = self._rhodecode_user.username != User.DEFAULT_USER
         if not allowed_user_search:
             return [], False
 
         name_contains = re.compile('(?:user_group:[ ]?)(.+)').findall(name_contains)
         if len(name_contains) != 1:
             return [], False
 
         name_contains = name_contains[0]
 
         query = UserGroup.query()\
             .order_by(func.length(UserGroup.users_group_name))\
             .order_by(UserGroup.users_group_name)
 
         if name_contains:
             ilike_expression = u'%{}%'.format(safe_unicode(name_contains))
             query = query.filter(
                 UserGroup.users_group_name.ilike(ilike_expression))
             query = query.limit(limit)
 
         acl_iter = query
 
         return [
             {
                 'id': obj.users_group_id,
                 'value': org_query,
                 'value_display': 'user_group: `{}`'.format(obj.users_group_name),
                 'type': 'user_group',
                 'url': h.route_path(
                     'user_group_profile', user_group_name=obj.users_group_name)
             }
             for obj in acl_iter], True
 
     def _get_hash_commit_list(self, auth_user, searcher, query, repo=None, repo_group=None):
         repo_name = repo_group_name = None
         if repo:
             repo_name = repo.repo_name
         if repo_group:
             repo_group_name = repo_group.group_name
 
         org_query = query
         if not query or len(query) < 3 or not searcher:
             return [], False
 
         commit_hashes = re.compile('(?:commit:[ ]?)([0-9a-f]{2,40})').findall(query)
 
         if len(commit_hashes) != 1:
             return [], False
 
         commit_hash = commit_hashes[0]
 
         result = searcher.search(
             'commit_id:{}*'.format(commit_hash), 'commit', auth_user,
             repo_name, repo_group_name, raise_on_exc=False)
 
         commits = []
         for entry in result['results']:
             repo_data = {
                 'repository_id': entry.get('repository_id'),
                 'repository_type': entry.get('repo_type'),
                 'repository_name': entry.get('repository'),
             }
 
             commit_entry = {
                 'id': entry['commit_id'],
                 'value': org_query,
                 'value_display': '`{}` commit: {}'.format(
                     entry['repository'], entry['commit_id']),
                 'type': 'commit',
                 'repo': entry['repository'],
                 'repo_data': repo_data,
 
                 'url': h.route_path(
                     'repo_commit',
                     repo_name=entry['repository'], commit_id=entry['commit_id'])
             }
 
             commits.append(commit_entry)
         return commits, True
 
     def _get_path_list(self, auth_user, searcher, query, repo=None, repo_group=None):
         repo_name = repo_group_name = None
         if repo:
             repo_name = repo.repo_name
         if repo_group:
             repo_group_name = repo_group.group_name
 
         org_query = query
         if not query or len(query) < 3 or not searcher:
             return [], False
 
         paths_re = re.compile('(?:file:[ ]?)(.+)').findall(query)
         if len(paths_re) != 1:
             return [], False
 
         file_path = paths_re[0]
 
         search_path = searcher.escape_specials(file_path)
         result = searcher.search(
             'file.raw:*{}*'.format(search_path), 'path', auth_user,
             repo_name, repo_group_name, raise_on_exc=False)
 
         files = []
         for entry in result['results']:
             repo_data = {
                 'repository_id': entry.get('repository_id'),
                 'repository_type': entry.get('repo_type'),
                 'repository_name': entry.get('repository'),
             }
 
             file_entry = {
                 'id': entry['commit_id'],
                 'value': org_query,
                 'value_display': '`{}` file: {}'.format(
                     entry['repository'], entry['file']),
                 'type': 'file',
                 'repo': entry['repository'],
                 'repo_data': repo_data,
 
                 'url': h.route_path(
                     'repo_files',
                     repo_name=entry['repository'], commit_id=entry['commit_id'],
                     f_path=entry['file'])
             }
 
             files.append(file_entry)
         return files, True
 
     @LoginRequired()
     @view_config(
         route_name='repo_list_data', request_method='GET',
         renderer='json_ext', xhr=True)
     def repo_list_data(self):
         _ = self.request.translate
         self.load_default_context()
 
         query = self.request.GET.get('query')
         repo_type = self.request.GET.get('repo_type')
         log.debug('generating repo list, query:%s, repo_type:%s',
                   query, repo_type)
 
         res = []
         repos = self._get_repo_list(query, repo_type=repo_type)
         if repos:
             res.append({
                 'text': _('Repositories'),
                 'children': repos
             })
 
         data = {
             'more': False,
             'results': res
         }
         return data
 
     @LoginRequired()
     @view_config(
         route_name='repo_group_list_data', request_method='GET',
         renderer='json_ext', xhr=True)
     def repo_group_list_data(self):
         _ = self.request.translate
         self.load_default_context()
 
         query = self.request.GET.get('query')
 
         log.debug('generating repo group list, query:%s',
                   query)
 
         res = []
         repo_groups = self._get_repo_group_list(query)
         if repo_groups:
             res.append({
                 'text': _('Repository Groups'),
                 'children': repo_groups
             })
 
         data = {
             'more': False,
             'results': res
         }
         return data
 
     def _get_default_search_queries(self, search_context, searcher, query):
         if not searcher:
             return []
 
         is_es_6 = searcher.is_es_6
 
         queries = []
         repo_group_name, repo_name, repo_context = None, None, None
 
         # repo group context
         if search_context.get('search_context[repo_group_name]'):
             repo_group_name = search_context.get('search_context[repo_group_name]')
         if search_context.get('search_context[repo_name]'):
             repo_name = search_context.get('search_context[repo_name]')
             repo_context = search_context.get('search_context[repo_view_type]')
 
         if is_es_6 and repo_name:
             # files
             def query_modifier():
                 qry = query
                 return {'q': qry, 'type': 'content'}
 
             label = u'File search for `{}`'.format(h.escape(query))
             file_qry = {
                 'id': -10,
                 'value': query,
                 'value_display': label,
                 'value_icon': '<i class="icon-code"></i>',
                 'type': 'search',
                 'subtype': 'repo',
                 'url': h.route_path('search_repo',
                                     repo_name=repo_name,
                                     _query=query_modifier())
                 }
 
             # commits
             def query_modifier():
                 qry = query
                 return {'q': qry, 'type': 'commit'}
 
             label = u'Commit search for `{}`'.format(h.escape(query))
             commit_qry = {
                 'id': -20,
                 'value': query,
                 'value_display': label,
                 'value_icon': '<i class="icon-history"></i>',
                 'type': 'search',
                 'subtype': 'repo',
                 'url': h.route_path('search_repo',
                                     repo_name=repo_name,
                                     _query=query_modifier())
                 }
 
             if repo_context in ['commit', 'commits']:
                 queries.extend([commit_qry, file_qry])
             elif repo_context in ['files', 'summary']:
                 queries.extend([file_qry, commit_qry])
             else:
                 queries.extend([commit_qry, file_qry])
 
         elif is_es_6 and repo_group_name:
             # files
             def query_modifier():
                 qry = query
                 return {'q': qry, 'type': 'content'}
 
             label = u'File search for `{}`'.format(query)
             file_qry = {
                 'id': -30,
                 'value': query,
                 'value_display': label,
                 'value_icon': '<i class="icon-code"></i>',
                 'type': 'search',
                 'subtype': 'repo_group',
                 'url': h.route_path('search_repo_group',
                                     repo_group_name=repo_group_name,
                                     _query=query_modifier())
                 }
 
             # commits
             def query_modifier():
                 qry = query
                 return {'q': qry, 'type': 'commit'}
 
             label = u'Commit search for `{}`'.format(query)
             commit_qry = {
                 'id': -40,
                 'value': query,
                 'value_display': label,
                 'value_icon': '<i class="icon-history"></i>',
                 'type': 'search',
                 'subtype': 'repo_group',
                 'url': h.route_path('search_repo_group',
                                     repo_group_name=repo_group_name,
                                     _query=query_modifier())
                 }
 
             if repo_context in ['commit', 'commits']:
                 queries.extend([commit_qry, file_qry])
             elif repo_context in ['files', 'summary']:
                 queries.extend([file_qry, commit_qry])
             else:
                 queries.extend([commit_qry, file_qry])
 
         # Global, not scoped
         if not queries:
             queries.append(
                 {
                     'id': -1,
                     'value': query,
                     'value_display': u'File search for: `{}`'.format(query),
                     'value_icon': '<i class="icon-code"></i>',
                     'type': 'search',
                     'subtype': 'global',
                     'url': h.route_path('search',
                                         _query={'q': query, 'type': 'content'})
                 })
             queries.append(
                 {
                     'id': -2,
                     'value': query,
                     'value_display': u'Commit search for: `{}`'.format(query),
                     'value_icon': '<i class="icon-history"></i>',
                     'type': 'search',
                     'subtype': 'global',
                     'url': h.route_path('search',
                                         _query={'q': query, 'type': 'commit'})
                 })
 
         return queries
 
     @LoginRequired()
     @view_config(
         route_name='goto_switcher_data', request_method='GET',
         renderer='json_ext', xhr=True)
     def goto_switcher_data(self):
         c = self.load_default_context()
 
         _ = self.request.translate
 
         query = self.request.GET.get('query')
         log.debug('generating main filter data, query %s', query)
 
         res = []
         if not query:
             return {'suggestions': res}
 
         def no_match(name):
             return {
                 'id': -1,
                 'value': "",
                 'value_display': name,
                 'type': 'text',
                 'url': ""
             }
         searcher = searcher_from_config(self.request.registry.settings)
         has_specialized_search = False
 
         # set repo context
         repo = None
         repo_id = safe_int(self.request.GET.get('search_context[repo_id]'))
         if repo_id:
             repo = Repository.get(repo_id)
 
         # set group context
         repo_group = None
         repo_group_id = safe_int(self.request.GET.get('search_context[repo_group_id]'))
         if repo_group_id:
             repo_group = RepoGroup.get(repo_group_id)
         prefix_match = False
 
         # user: type search
         if not prefix_match:
             users, prefix_match = self._get_user_list(query)
             if users:
                 has_specialized_search = True
                 for serialized_user in users:
                     res.append(serialized_user)
             elif prefix_match:
                 has_specialized_search = True
                 res.append(no_match('No matching users found'))
 
         # user_group: type search
         if not prefix_match:
             user_groups, prefix_match = self._get_user_groups_list(query)
             if user_groups:
                 has_specialized_search = True
                 for serialized_user_group in user_groups:
                     res.append(serialized_user_group)
             elif prefix_match:
                 has_specialized_search = True
                 res.append(no_match('No matching user groups found'))
 
         # FTS commit: type search
         if not prefix_match:
             commits, prefix_match = self._get_hash_commit_list(
                 c.auth_user, searcher, query, repo, repo_group)
             if commits:
                 has_specialized_search = True
                 unique_repos = collections.OrderedDict()
                 for commit in commits:
                     repo_name = commit['repo']
                     unique_repos.setdefault(repo_name, []).append(commit)
 
                 for _repo, commits in unique_repos.items():
                     for commit in commits:
                         res.append(commit)
             elif prefix_match:
                 has_specialized_search = True
                 res.append(no_match('No matching commits found'))
 
         # FTS file: type search
         if not prefix_match:
             paths, prefix_match = self._get_path_list(
                 c.auth_user, searcher, query, repo, repo_group)
             if paths:
                 has_specialized_search = True
                 unique_repos = collections.OrderedDict()
                 for path in paths:
                     repo_name = path['repo']
                     unique_repos.setdefault(repo_name, []).append(path)
 
                 for repo, paths in unique_repos.items():
                     for path in paths:
                         res.append(path)
             elif prefix_match:
                 has_specialized_search = True
                 res.append(no_match('No matching files found'))
 
         # main suggestions
         if not has_specialized_search:
             repo_group_name = ''
             if repo_group:
                 repo_group_name = repo_group.group_name
 
             for _q in self._get_default_search_queries(self.request.GET, searcher, query):
                 res.append(_q)
 
             repo_groups = self._get_repo_group_list(query, repo_group_name=repo_group_name)
             for serialized_repo_group in repo_groups:
                 res.append(serialized_repo_group)
 
             repos = self._get_repo_list(query, repo_group_name=repo_group_name)
             for serialized_repo in repos:
                 res.append(serialized_repo)
 
             if not repos and not repo_groups:
                 res.append(no_match('No matches found'))
 
         return {'suggestions': res}
 
     @LoginRequired()
     @view_config(
         route_name='home', request_method='GET',
         renderer='rhodecode:templates/index.mako')
     def main_page(self):
         c = self.load_default_context()
         c.repo_group = None
         return self._get_template_context(c)
 
     def _main_page_repo_groups_data(self, repo_group_id):
         column_map = {
             'name': 'group_name_hash',
             'desc': 'group_description',
             'last_change': 'updated_on',
             'owner': 'user_username',
         }
         draw, start, limit = self._extract_chunk(self.request)
         search_q, order_by, order_dir = self._extract_ordering(
             self.request, column_map=column_map)
         return RepoGroupModel().get_repo_groups_data_table(
             draw, start, limit,
             search_q, order_by, order_dir,
             self._rhodecode_user, repo_group_id)
 
     def _main_page_repos_data(self, repo_group_id):
         column_map = {
             'name': 'repo_name',
             'desc': 'description',
             'last_change': 'updated_on',
             'owner': 'user_username',
         }
         draw, start, limit = self._extract_chunk(self.request)
         search_q, order_by, order_dir = self._extract_ordering(
             self.request, column_map=column_map)
         return RepoModel().get_repos_data_table(
             draw, start, limit,
             search_q, order_by, order_dir,
             self._rhodecode_user, repo_group_id)
 
     @LoginRequired()
     @view_config(
         route_name='main_page_repo_groups_data',
         request_method='GET', renderer='json_ext', xhr=True)
     def main_page_repo_groups_data(self):
         self.load_default_context()
         repo_group_id = safe_int(self.request.GET.get('repo_group_id'))
 
         if repo_group_id:
             group = RepoGroup.get_or_404(repo_group_id)
-            _perms = ['group.read', 'group.write', 'group.admin']
+            _perms = AuthUser.repo_group_read_perms
             if not HasRepoGroupPermissionAny(*_perms)(
                     group.group_name, 'user is allowed to list repo group children'):
                 raise HTTPNotFound()
 
         return self._main_page_repo_groups_data(repo_group_id)
 
     @LoginRequired()
     @view_config(
         route_name='main_page_repos_data',
         request_method='GET', renderer='json_ext', xhr=True)
     def main_page_repos_data(self):
         self.load_default_context()
         repo_group_id = safe_int(self.request.GET.get('repo_group_id'))
 
         if repo_group_id:
             group = RepoGroup.get_or_404(repo_group_id)
-            _perms = ['group.read', 'group.write', 'group.admin']
+            _perms = AuthUser.repo_group_read_perms
             if not HasRepoGroupPermissionAny(*_perms)(
                     group.group_name, 'user is allowed to list repo group children'):
                 raise HTTPNotFound()
 
         return self._main_page_repos_data(repo_group_id)
 
     @LoginRequired()
-    @HasRepoGroupPermissionAnyDecorator(
-        'group.read', 'group.write', 'group.admin')
+    @HasRepoGroupPermissionAnyDecorator(*AuthUser.repo_group_read_perms)
     @view_config(
         route_name='repo_group_home', request_method='GET',
         renderer='rhodecode:templates/index_repo_group.mako')
     @view_config(
         route_name='repo_group_home_slash', request_method='GET',
         renderer='rhodecode:templates/index_repo_group.mako')
     def repo_group_main_page(self):
         c = self.load_default_context()
         c.repo_group = self.request.db_repo_group
         return self._get_template_context(c)
 
     @LoginRequired()
     @CSRFRequired()
     @view_config(
         route_name='markup_preview', request_method='POST',
         renderer='string', xhr=True)
     def markup_preview(self):
         # Technically a CSRF token is not needed as no state changes with this
         # call. However, as this is a POST is better to have it, so automated
         # tools don't flag it as potential CSRF.
         # Post is required because the payload could be bigger than the maximum
         # allowed by GET.
 
         text = self.request.POST.get('text')
         renderer = self.request.POST.get('renderer') or 'rst'
         if text:
             return h.render(text, renderer=renderer, mentions=True)
         return ''
 
     @LoginRequired()
     @CSRFRequired()
     @view_config(
         route_name='file_preview', request_method='POST',
         renderer='string', xhr=True)
     def file_preview(self):
         # Technically a CSRF token is not needed as no state changes with this
         # call. However, as this is a POST is better to have it, so automated
         # tools don't flag it as potential CSRF.
         # Post is required because the payload could be bigger than the maximum
         # allowed by GET.
 
         text = self.request.POST.get('text')
         file_path = self.request.POST.get('file_path')
 
         renderer = h.renderer_from_filename(file_path)
 
         if renderer:
             return h.render(text, renderer=renderer, mentions=True)
         else:
             self.load_default_context()
             _render = self.request.get_partial_renderer(
                 'rhodecode:templates/files/file_content.mako')
 
             lines = filenode_as_lines_tokens(FileNode(file_path, text))
 
             return _render('render_lines', lines)
 
     @LoginRequired()
     @CSRFRequired()
     @view_config(
         route_name='store_user_session_value', request_method='POST',
         renderer='string', xhr=True)
     def store_user_session_attr(self):
         key = self.request.POST.get('key')
         val = self.request.POST.get('val')
 
         existing_value = self.request.session.get(key)
         if existing_value != val:
             self.request.session[key] = val
 
         return 'stored:{}:{}'.format(key, val)

rhodecode/apps/login/tests/test_login.py

@@ -1,579 +1,578 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2019 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 urlparse
 
 import mock
 import pytest
 
 from rhodecode.tests import (
     assert_session_flash, HG_REPO, TEST_USER_ADMIN_LOGIN,
     no_newline_id_generator)
 from rhodecode.tests.fixture import Fixture
 from rhodecode.lib.auth import check_password
 from rhodecode.lib import helpers as h
 from rhodecode.model.auth_token import AuthTokenModel
 from rhodecode.model.db import User, Notification, UserApiKeys
 from rhodecode.model.meta import Session
 
 fixture = Fixture()
 
 whitelist_view = ['RepoCommitsView:repo_commit_raw']
 
 
 def route_path(name, params=None, **kwargs):
     import urllib
     from rhodecode.apps._base import ADMIN_PREFIX
 
     base_url = {
         'login': ADMIN_PREFIX + '/login',
         'logout': ADMIN_PREFIX + '/logout',
         'register': ADMIN_PREFIX + '/register',
         'reset_password':
             ADMIN_PREFIX + '/password_reset',
         'reset_password_confirmation':
             ADMIN_PREFIX + '/password_reset_confirmation',
 
         'admin_permissions_application':
             ADMIN_PREFIX + '/permissions/application',
         'admin_permissions_application_update':
             ADMIN_PREFIX + '/permissions/application/update',
 
         'repo_commit_raw': '/{repo_name}/raw-changeset/{commit_id}'
 
     }[name].format(**kwargs)
 
     if params:
         base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
     return base_url
 
 
 @pytest.mark.usefixtures('app')
 class TestLoginController(object):
     destroy_users = set()
 
     @classmethod
     def teardown_class(cls):
         fixture.destroy_users(cls.destroy_users)
 
     def teardown_method(self, method):
         for n in Notification.query().all():
             Session().delete(n)
 
         Session().commit()
         assert Notification.query().all() == []
 
     def test_index(self):
         response = self.app.get(route_path('login'))
         assert response.status == '200 OK'
         # Test response...
 
     def test_login_admin_ok(self):
         response = self.app.post(route_path('login'),
                                  {'username': 'test_admin',
                                   'password': 'test12'}, status=302)
         response = response.follow()
         session = response.get_session_from_response()
         username = session['rhodecode_user'].get('username')
         assert username == 'test_admin'
-        response.mustcontain('/%s' % HG_REPO)
+        response.mustcontain('logout')
 
     def test_login_regular_ok(self):
         response = self.app.post(route_path('login'),
                                  {'username': 'test_regular',
                                   'password': 'test12'}, status=302)
 
         response = response.follow()
         session = response.get_session_from_response()
         username = session['rhodecode_user'].get('username')
         assert username == 'test_regular'
-
-        response.mustcontain('/%s' % HG_REPO)
+        response.mustcontain('logout')
 
     def test_login_regular_forbidden_when_super_admin_restriction(self):
         from rhodecode.authentication.plugins.auth_rhodecode import RhodeCodeAuthPlugin
         with fixture.auth_restriction(RhodeCodeAuthPlugin.AUTH_RESTRICTION_SUPER_ADMIN):
             response = self.app.post(route_path('login'),
                                      {'username': 'test_regular',
                                       'password': 'test12'})
 
             response.mustcontain('invalid user name')
             response.mustcontain('invalid password')
 
     def test_login_regular_forbidden_when_scope_restriction(self):
         from rhodecode.authentication.plugins.auth_rhodecode import RhodeCodeAuthPlugin
         with fixture.scope_restriction(RhodeCodeAuthPlugin.AUTH_RESTRICTION_SCOPE_VCS):
             response = self.app.post(route_path('login'),
                                      {'username': 'test_regular',
                                       'password': 'test12'})
 
             response.mustcontain('invalid user name')
             response.mustcontain('invalid password')
 
     def test_login_ok_came_from(self):
         test_came_from = '/_admin/users?branch=stable'
         _url = '{}?came_from={}'.format(route_path('login'), test_came_from)
         response = self.app.post(
             _url, {'username': 'test_admin', 'password': 'test12'}, status=302)
 
         assert 'branch=stable' in response.location
         response = response.follow()
 
         assert response.status == '200 OK'
         response.mustcontain('Users administration')
 
     def test_redirect_to_login_with_get_args(self):
         with fixture.anon_access(False):
             kwargs = {'branch': 'stable'}
             response = self.app.get(
                 h.route_path('repo_summary', repo_name=HG_REPO, _query=kwargs),
                 status=302)
 
             response_query = urlparse.parse_qsl(response.location)
             assert 'branch=stable' in response_query[0][1]
 
     def test_login_form_with_get_args(self):
         _url = '{}?came_from=/_admin/users,branch=stable'.format(route_path('login'))
         response = self.app.get(_url)
         assert 'branch%3Dstable' in response.form.action
 
     @pytest.mark.parametrize("url_came_from", [
         'data:text/html,<script>window.alert("xss")</script>',
         'mailto:[e-mail]',
         'file:///etc/passwd',
         'ftp://some.ftp.server',
         'http://other.domain',
         '/\r\nX-Forwarded-Host: http://example.org',
     ], ids=no_newline_id_generator)
     def test_login_bad_came_froms(self, url_came_from):
         _url = '{}?came_from={}'.format(route_path('login'), url_came_from)
         response = self.app.post(
             _url,
             {'username': 'test_admin', 'password': 'test12'})
         assert response.status == '302 Found'
         response = response.follow()
         assert response.status == '200 OK'
         assert response.request.path == '/'
 
     def test_login_short_password(self):
         response = self.app.post(route_path('login'),
                                  {'username': 'test_admin',
                                   'password': 'as'})
         assert response.status == '200 OK'
 
         response.mustcontain('Enter 3 characters or more')
 
     def test_login_wrong_non_ascii_password(self, user_regular):
         response = self.app.post(
             route_path('login'),
             {'username': user_regular.username,
              'password': u'invalid-non-asci\xe4'.encode('utf8')})
 
         response.mustcontain('invalid user name')
         response.mustcontain('invalid password')
 
     def test_login_with_non_ascii_password(self, user_util):
         password = u'valid-non-ascii\xe4'
         user = user_util.create_user(password=password)
         response = self.app.post(
             route_path('login'),
             {'username': user.username,
              'password': password.encode('utf-8')})
         assert response.status_code == 302
 
     def test_login_wrong_username_password(self):
         response = self.app.post(route_path('login'),
                                  {'username': 'error',
                                   'password': 'test12'})
 
         response.mustcontain('invalid user name')
         response.mustcontain('invalid password')
 
     def test_login_admin_ok_password_migration(self, real_crypto_backend):
         from rhodecode.lib import auth
 
         # create new user, with sha256 password
         temp_user = 'test_admin_sha256'
         user = fixture.create_user(temp_user)
         user.password = auth._RhodeCodeCryptoSha256().hash_create(
             b'test123')
         Session().add(user)
         Session().commit()
         self.destroy_users.add(temp_user)
         response = self.app.post(route_path('login'),
                                  {'username': temp_user,
                                   'password': 'test123'}, status=302)
 
         response = response.follow()
         session = response.get_session_from_response()
         username = session['rhodecode_user'].get('username')
         assert username == temp_user
-        response.mustcontain('/%s' % HG_REPO)
+        response.mustcontain('logout')
 
         # new password should be bcrypted, after log-in and transfer
         user = User.get_by_username(temp_user)
         assert user.password.startswith('$')
 
     # REGISTRATIONS
     def test_register(self):
         response = self.app.get(route_path('register'))
         response.mustcontain('Create an Account')
 
     def test_register_err_same_username(self):
         uname = 'test_admin'
         response = self.app.post(
             route_path('register'),
             {
                 'username': uname,
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         assertr = response.assert_response()
         msg = 'Username "%(username)s" already exists'
         msg = msg % {'username': uname}
         assertr.element_contains('#username+.error-message', msg)
 
     def test_register_err_same_email(self):
         response = self.app.post(
             route_path('register'),
             {
                 'username': 'test_admin_0',
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         assertr = response.assert_response()
         msg = u'This e-mail address is already taken'
         assertr.element_contains('#email+.error-message', msg)
 
     def test_register_err_same_email_case_sensitive(self):
         response = self.app.post(
             route_path('register'),
             {
                 'username': 'test_admin_1',
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
         assertr = response.assert_response()
         msg = u'This e-mail address is already taken'
         assertr.element_contains('#email+.error-message', msg)
 
     def test_register_err_wrong_data(self):
         response = self.app.post(
             route_path('register'),
             {
                 'username': 'xs',
                 'password': 'test',
                 'password_confirmation': 'test',
                 'email': 'goodmailm',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
         assert response.status == '200 OK'
         response.mustcontain('An email address must contain a single @')
         response.mustcontain('Enter a value 6 characters long or more')
 
     def test_register_err_username(self):
         response = self.app.post(
             route_path('register'),
             {
                 'username': 'error user',
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': 'goodmailm',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         response.mustcontain('An email address must contain a single @')
         response.mustcontain(
             'Username may only contain '
             'alphanumeric characters underscores, '
             'periods or dashes and must begin with '
             'alphanumeric character')
 
     def test_register_err_case_sensitive(self):
         usr = 'Test_Admin'
         response = self.app.post(
             route_path('register'),
             {
                 'username': usr,
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': 'goodmailm',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         assertr = response.assert_response()
         msg = u'Username "%(username)s" already exists'
         msg = msg % {'username': usr}
         assertr.element_contains('#username+.error-message', msg)
 
     def test_register_special_chars(self):
         response = self.app.post(
             route_path('register'),
             {
                 'username': 'xxxaxn',
                 'password': 'ąćźżąśśśś',
                 'password_confirmation': 'ąćźżąśśśś',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         msg = u'Invalid characters (non-ascii) in password'
         response.mustcontain(msg)
 
     def test_register_password_mismatch(self):
         response = self.app.post(
             route_path('register'),
             {
                 'username': 'xs',
                 'password': '123qwe',
                 'password_confirmation': 'qwe123',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
         msg = u'Passwords do not match'
         response.mustcontain(msg)
 
     def test_register_ok(self):
         username = 'test_regular4'
         password = 'qweqwe'
         email = '[e-mail]'
         name = 'testname'
         lastname = 'testlastname'
 
         # this initializes a session
         response = self.app.get(route_path('register'))
         response.mustcontain('Create an Account')
 
 
         response = self.app.post(
             route_path('register'),
             {
                 'username': username,
                 'password': password,
                 'password_confirmation': password,
                 'email': email,
                 'firstname': name,
                 'lastname': lastname,
                 'admin': True
             },
             status=302
         )  # This should be overridden
 
         assert_session_flash(
             response, 'You have successfully registered with RhodeCode. You can log-in now.')
 
         ret = Session().query(User).filter(
             User.username == 'test_regular4').one()
         assert ret.username == username
         assert check_password(password, ret.password)
         assert ret.email == email
         assert ret.name == name
         assert ret.lastname == lastname
         assert ret.auth_tokens is not None
         assert not ret.admin
 
     def test_forgot_password_wrong_mail(self):
         bad_email = '[e-mail]'
         # this initializes a session
         self.app.get(route_path('reset_password'))
 
         response = self.app.post(
             route_path('reset_password'), {'email': bad_email, }
         )
         assert_session_flash(response,
             'If such email exists, a password reset link was sent to it.')
 
     def test_forgot_password(self, user_util):
         # this initializes a session
         self.app.get(route_path('reset_password'))
 
         user = user_util.create_user()
         user_id = user.user_id
         email = user.email
 
         response = self.app.post(route_path('reset_password'), {'email': email, })
 
         assert_session_flash(response,
             'If such email exists, a password reset link was sent to it.')
 
         # BAD KEY
         confirm_url = '{}?key={}'.format(route_path('reset_password_confirmation'), 'badkey')
         response = self.app.get(confirm_url, status=302)
         assert response.location.endswith(route_path('reset_password'))
         assert_session_flash(response, 'Given reset token is invalid')
 
         response.follow()  # cleanup flash
 
         # GOOD KEY
         key = UserApiKeys.query()\
             .filter(UserApiKeys.user_id == user_id)\
             .filter(UserApiKeys.role == UserApiKeys.ROLE_PASSWORD_RESET)\
             .first()
 
         assert key
 
         confirm_url = '{}?key={}'.format(route_path('reset_password_confirmation'), key.api_key)
         response = self.app.get(confirm_url)
         assert response.status == '302 Found'
         assert response.location.endswith(route_path('login'))
 
         assert_session_flash(
             response,
             'Your password reset was successful, '
             'a new password has been sent to your email')
 
         response.follow()
 
     def _get_api_whitelist(self, values=None):
         config = {'api_access_controllers_whitelist': values or []}
         return config
 
     @pytest.mark.parametrize("test_name, auth_token", [
         ('none', None),
         ('empty_string', ''),
         ('fake_number', '123456'),
         ('proper_auth_token', None)
     ])
     def test_access_not_whitelisted_page_via_auth_token(
             self, test_name, auth_token, user_admin):
 
         whitelist = self._get_api_whitelist([])
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
             assert [] == whitelist['api_access_controllers_whitelist']
             if test_name == 'proper_auth_token':
                 # use builtin if api_key is None
                 auth_token = user_admin.api_key
 
             with fixture.anon_access(False):
                 self.app.get(
                     route_path('repo_commit_raw',
                                repo_name=HG_REPO, commit_id='tip',
                                params=dict(api_key=auth_token)),
                     status=302)
 
     @pytest.mark.parametrize("test_name, auth_token, code", [
         ('none', None, 302),
         ('empty_string', '', 302),
         ('fake_number', '123456', 302),
         ('proper_auth_token', None, 200)
     ])
     def test_access_whitelisted_page_via_auth_token(
             self, test_name, auth_token, code, user_admin):
 
         whitelist = self._get_api_whitelist(whitelist_view)
 
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
             assert whitelist_view == whitelist['api_access_controllers_whitelist']
 
             if test_name == 'proper_auth_token':
                 auth_token = user_admin.api_key
                 assert auth_token
 
             with fixture.anon_access(False):
                 self.app.get(
                     route_path('repo_commit_raw',
                                repo_name=HG_REPO, commit_id='tip',
                                params=dict(api_key=auth_token)),
                     status=code)
 
     @pytest.mark.parametrize("test_name, auth_token, code", [
         ('proper_auth_token', None, 200),
         ('wrong_auth_token', '123456', 302),
     ])
     def test_access_whitelisted_page_via_auth_token_bound_to_token(
             self, test_name, auth_token, code, user_admin):
 
         expected_token = auth_token
         if test_name == 'proper_auth_token':
             auth_token = user_admin.api_key
             expected_token = auth_token
             assert auth_token
 
         whitelist = self._get_api_whitelist([
             'RepoCommitsView:repo_commit_raw@{}'.format(expected_token)])
 
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
 
             with fixture.anon_access(False):
                 self.app.get(
                     route_path('repo_commit_raw',
                                repo_name=HG_REPO, commit_id='tip',
                                params=dict(api_key=auth_token)),
                     status=code)
 
     def test_access_page_via_extra_auth_token(self):
         whitelist = self._get_api_whitelist(whitelist_view)
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
             assert whitelist_view == \
                 whitelist['api_access_controllers_whitelist']
 
             new_auth_token = AuthTokenModel().create(
                 TEST_USER_ADMIN_LOGIN, 'test')
             Session().commit()
             with fixture.anon_access(False):
                 self.app.get(
                     route_path('repo_commit_raw',
                                repo_name=HG_REPO, commit_id='tip',
                                params=dict(api_key=new_auth_token.api_key)),
                     status=200)
 
     def test_access_page_via_expired_auth_token(self):
         whitelist = self._get_api_whitelist(whitelist_view)
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
             assert whitelist_view == \
                 whitelist['api_access_controllers_whitelist']
 
             new_auth_token = AuthTokenModel().create(
                 TEST_USER_ADMIN_LOGIN, 'test')
             Session().commit()
             # patch the api key and make it expired
             new_auth_token.expires = 0
             Session().add(new_auth_token)
             Session().commit()
             with fixture.anon_access(False):
                 self.app.get(
                     route_path('repo_commit_raw',
                                repo_name=HG_REPO, commit_id='tip',
                                params=dict(api_key=new_auth_token.api_key)),
                     status=302)

rhodecode/tests/functional/test_delegated_admin.py

@@ -1,152 +1,157 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2019 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
 
 
 def route_path(name, params=None, **kwargs):
     import urllib
     from rhodecode.apps._base import ADMIN_PREFIX
 
     base_url = {
         'home': '/',
         'admin_home': ADMIN_PREFIX,
         'repos':
             ADMIN_PREFIX + '/repos',
+        'repos_data':
+            ADMIN_PREFIX + '/repos_data',
         'repo_groups':
             ADMIN_PREFIX + '/repo_groups',
         'repo_groups_data':
             ADMIN_PREFIX + '/repo_groups_data',
         'user_groups':
             ADMIN_PREFIX + '/user_groups',
         'user_groups_data':
             ADMIN_PREFIX + '/user_groups_data',
     }[name].format(**kwargs)
 
     if params:
         base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
     return base_url
 
 
 fixture = Fixture()
 
 
 class TestAdminDelegatedUser(TestController):
 
     def test_regular_user_cannot_see_admin_interfaces(self, user_util, xhr_header):
         user = user_util.create_user(password='qweqwe')
         user_util.inherit_default_user_permissions(user.username, False)
 
         self.log_user(user.username, 'qweqwe')
 
         # user doesn't have any access to resources so main admin page should 404
         self.app.get(route_path('admin_home'), status=404)
 
-        response = self.app.get(route_path('repos'), status=200)
-        response.mustcontain('data: []')
+        response = self.app.get(route_path('repos_data'),
+                                status=200, extra_environ=xhr_header)
+        assert response.json['data'] == []
 
         response = self.app.get(route_path('repo_groups_data'),
                                 status=200, extra_environ=xhr_header)
         assert response.json['data'] == []
 
         response = self.app.get(route_path('user_groups_data'),
                                 status=200, extra_environ=xhr_header)
         assert response.json['data'] == []
 
     def test_regular_user_can_see_admin_interfaces_if_owner(self, user_util, xhr_header):
         user = user_util.create_user(password='qweqwe')
         username = user.username
 
         repo = user_util.create_repo(owner=username)
         repo_name = repo.repo_name
 
         repo_group = user_util.create_repo_group(owner=username)
         repo_group_name = repo_group.group_name
 
         user_group = user_util.create_user_group(owner=username)
         user_group_name = user_group.users_group_name
 
         self.log_user(username, 'qweqwe')
 
         response = self.app.get(route_path('admin_home'))
 
         assert_response = response.assert_response()
 
         assert_response.element_contains('td.delegated-admin-repos', '1')
         assert_response.element_contains('td.delegated-admin-repo-groups', '1')
         assert_response.element_contains('td.delegated-admin-user-groups', '1')
 
         # admin interfaces have visible elements
-        response = self.app.get(route_path('repos'), status=200)
-        response.mustcontain('"name_raw": "{}"'.format(repo_name))
+        response = self.app.get(route_path('repos_data'),
+                                extra_environ=xhr_header, status=200)
+        response.mustcontain('<a href=\\"/{}\\">'.format(repo_name))
 
         response = self.app.get(route_path('repo_groups_data'),
                                 extra_environ=xhr_header, status=200)
-        response.mustcontain('"name_raw": "{}"'.format(repo_group_name))
+        response.mustcontain('<a href=\\"/{}\\">'.format(repo_group_name))
 
         response = self.app.get(route_path('user_groups_data'),
                                 extra_environ=xhr_header, status=200)
-        response.mustcontain('"name_raw": "{}"'.format(user_group_name))
+        response.mustcontain('<a href=\\"/_profile_user_group/{}\\">'.format(user_group_name))
 
     def test_regular_user_can_see_admin_interfaces_if_admin_perm(
             self, user_util, xhr_header):
         user = user_util.create_user(password='qweqwe')
         username = user.username
 
         repo = user_util.create_repo()
         repo_name = repo.repo_name
 
         repo_group = user_util.create_repo_group()
         repo_group_name = repo_group.group_name
 
         user_group = user_util.create_user_group()
         user_group_name = user_group.users_group_name
 
         user_util.grant_user_permission_to_repo(
             repo, user, 'repository.admin')
         user_util.grant_user_permission_to_repo_group(
             repo_group, user, 'group.admin')
         user_util.grant_user_permission_to_user_group(
             user_group, user, 'usergroup.admin')
 
         self.log_user(username, 'qweqwe')
         # check if in home view, such user doesn't see the "admin" menus
         response = self.app.get(route_path('admin_home'))
 
         assert_response = response.assert_response()
 
         assert_response.element_contains('td.delegated-admin-repos', '1')
         assert_response.element_contains('td.delegated-admin-repo-groups', '1')
         assert_response.element_contains('td.delegated-admin-user-groups', '1')
 
         # admin interfaces have visible elements
-        response = self.app.get(route_path('repos'), status=200)
-        response.mustcontain('"name_raw": "{}"'.format(repo_name))
+        response = self.app.get(route_path('repos_data'),
+                                extra_environ=xhr_header, status=200)
+        response.mustcontain('<a href=\\"/{}\\">'.format(repo_name))
 
         response = self.app.get(route_path('repo_groups_data'),
                                 extra_environ=xhr_header, status=200)
-        response.mustcontain('"name_raw": "{}"'.format(repo_group_name))
+        response.mustcontain('<a href=\\"/{}\\">'.format(repo_group_name))
 
         response = self.app.get(route_path('user_groups_data'),
                                 extra_environ=xhr_header, status=200)
-        response.mustcontain('"name_raw": "{}"'.format(user_group_name))
+        response.mustcontain('<a href=\\"/_profile_user_group/{}\\">'.format(user_group_name))