chore(cleanups): cleanup unicode usage...

rhodecode/api/tests/test_create_repo.py

@@ -1,348 +1,348 b''
 
 # Copyright (C) 2010-2023 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 mock
 import pytest
 
 from rhodecode.lib.vcs import settings
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.user import UserModel
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error, crash)
 from rhodecode.tests.fixture import Fixture
 from rhodecode.lib.ext_json import json
 from rhodecode.lib.str_utils import safe_str
 
 
 fixture = Fixture()
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestCreateRepo(object):
 
     @pytest.mark.parametrize('given, expected_name, expected_exc', [
         ('api repo-1', 'api-repo-1', False),
         ('api-repo 1-ąć', 'api-repo-1-ąć', False),
-        (u'unicode-ąć', u'unicode-ąć', False),
+        ('unicode-ąć', u'unicode-ąć', False),
         ('some repo v1.2', 'some-repo-v1.2', False),
         ('v2.0', 'v2.0', False),
     ])
     def test_api_create_repo(self, backend, given, expected_name, expected_exc):
 
         id_, params = build_data(
             self.apikey,
             'create_repo',
             repo_name=given,
             owner=TEST_USER_ADMIN_LOGIN,
             repo_type=backend.alias,
         )
         response = api_call(self.app, params)
 
         ret = {
             'msg': 'Created new repository `%s`' % (expected_name,),
             'success': True,
             'task': None,
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
         repo = RepoModel().get_by_repo_name(safe_str(expected_name))
         assert repo is not None
 
         id_, params = build_data(self.apikey, 'get_repo', repoid=expected_name)
         response = api_call(self.app, params)
         body = json.loads(response.body)
 
         assert body['result']['enable_downloads'] is False
         assert body['result']['enable_locking'] is False
         assert body['result']['enable_statistics'] is False
 
         fixture.destroy_repo(safe_str(expected_name))
 
     def test_api_create_restricted_repo_type(self, backend):
         repo_name = 'api-repo-type-{0}'.format(backend.alias)
         id_, params = build_data(
             self.apikey,
             'create_repo',
             repo_name=repo_name,
             owner=TEST_USER_ADMIN_LOGIN,
             repo_type=backend.alias,
         )
         git_backend = settings.BACKENDS['git']
         with mock.patch(
                 'rhodecode.lib.vcs.settings.BACKENDS', {'git': git_backend}):
             response = api_call(self.app, params)
 
         repo = RepoModel().get_by_repo_name(repo_name)
 
         if backend.alias == 'git':
             assert repo is not None
             expected = {
                 'msg': 'Created new repository `{0}`'.format(repo_name,),
                 'success': True,
                 'task': None,
             }
             assert_ok(id_, expected, given=response.body)
         else:
             assert repo is None
 
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_with_booleans(self, backend):
         repo_name = 'api-repo-2'
         id_, params = build_data(
             self.apikey,
             'create_repo',
             repo_name=repo_name,
             owner=TEST_USER_ADMIN_LOGIN,
             repo_type=backend.alias,
             enable_statistics=True,
             enable_locking=True,
             enable_downloads=True
         )
         response = api_call(self.app, params)
 
         repo = RepoModel().get_by_repo_name(repo_name)
 
         assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % (repo_name,),
             'success': True,
             'task': None,
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
 
         id_, params = build_data(self.apikey, 'get_repo', repoid=repo_name)
         response = api_call(self.app, params)
         body = json.loads(response.body)
 
         assert body['result']['enable_downloads'] is True
         assert body['result']['enable_locking'] is True
         assert body['result']['enable_statistics'] is True
 
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_in_group(self, backend):
         repo_group_name = 'my_gr'
         # create the parent
         fixture.create_repo_group(repo_group_name)
 
         repo_name = '%s/api-repo-gr' % (repo_group_name,)
         id_, params = build_data(
             self.apikey, 'create_repo',
             repo_name=repo_name,
             owner=TEST_USER_ADMIN_LOGIN,
             repo_type=backend.alias,)
         response = api_call(self.app, params)
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo is not None
         assert repo.group is not None
 
         ret = {
             'msg': 'Created new repository `%s`' % (repo_name,),
             'success': True,
             'task': None,
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
         fixture.destroy_repo_group(repo_group_name)
 
     def test_create_repo_in_group_that_doesnt_exist(self, backend, user_util):
         repo_group_name = 'fake_group'
 
         repo_name = '%s/api-repo-gr' % (repo_group_name,)
         id_, params = build_data(
             self.apikey, 'create_repo',
             repo_name=repo_name,
             owner=TEST_USER_ADMIN_LOGIN,
             repo_type=backend.alias,)
         response = api_call(self.app, params)
 
         expected = {'repo_group': 'Repository group `{}` does not exist'.format(
             repo_group_name)}
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_repo_unknown_owner(self, backend):
         repo_name = 'api-repo-2'
         owner = 'i-dont-exist'
         id_, params = build_data(
             self.apikey, 'create_repo',
             repo_name=repo_name,
             owner=owner,
             repo_type=backend.alias)
         response = api_call(self.app, params)
         expected = 'user `%s` does not exist' % (owner,)
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_repo_dont_specify_owner(self, backend):
         repo_name = 'api-repo-3'
         id_, params = build_data(
             self.apikey, 'create_repo',
             repo_name=repo_name,
             repo_type=backend.alias)
         response = api_call(self.app, params)
 
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % (repo_name,),
             'success': True,
             'task': None,
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_by_non_admin(self, backend):
         repo_name = 'api-repo-4'
         id_, params = build_data(
             self.apikey_regular, 'create_repo',
             repo_name=repo_name,
             repo_type=backend.alias)
         response = api_call(self.app, params)
 
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % (repo_name,),
             'success': True,
             'task': None,
         }
         expected = ret
         assert_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_by_non_admin_specify_owner(self, backend):
         repo_name = 'api-repo-5'
         owner = 'i-dont-exist'
         id_, params = build_data(
             self.apikey_regular, 'create_repo',
             repo_name=repo_name,
             repo_type=backend.alias,
             owner=owner)
         response = api_call(self.app, params)
 
         expected = 'Only RhodeCode super-admin can specify `owner` param'
         assert_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_by_non_admin_no_parent_group_perms(self, backend):
         repo_group_name = 'no-access'
         fixture.create_repo_group(repo_group_name)
         repo_name = 'no-access/api-repo'
 
         id_, params = build_data(
             self.apikey_regular, 'create_repo',
             repo_name=repo_name,
             repo_type=backend.alias)
         response = api_call(self.app, params)
 
         expected = {'repo_group': 'Repository group `{}` does not exist'.format(
             repo_group_name)}
         assert_error(id_, expected, given=response.body)
         fixture.destroy_repo_group(repo_group_name)
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_non_admin_no_permission_to_create_to_root_level(
             self, backend, user_util):
 
         regular_user = user_util.create_user()
         regular_user_api_key = regular_user.api_key
 
         usr = UserModel().get_by_username(regular_user.username)
         usr.inherit_default_permissions = False
         Session().add(usr)
 
         repo_name = backend.new_repo_name()
         id_, params = build_data(
             regular_user_api_key, 'create_repo',
             repo_name=repo_name,
             repo_type=backend.alias)
         response = api_call(self.app, params)
         expected = {
             "repo_name": "You do not have the permission to "
                          "store repositories in the root location."}
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_repo_exists(self, backend):
         repo_name = backend.repo_name
         id_, params = build_data(
             self.apikey, 'create_repo',
             repo_name=repo_name,
             owner=TEST_USER_ADMIN_LOGIN,
             repo_type=backend.alias,)
         response = api_call(self.app, params)
         expected = {
             'unique_repo_name': 'Repository with name `{}` already exists'.format(
                 repo_name)}
         assert_error(id_, expected, given=response.body)
 
     @mock.patch.object(RepoModel, 'create', crash)
     def test_api_create_repo_exception_occurred(self, backend):
         repo_name = 'api-repo-6'
         id_, params = build_data(
             self.apikey, 'create_repo',
             repo_name=repo_name,
             owner=TEST_USER_ADMIN_LOGIN,
             repo_type=backend.alias,)
         response = api_call(self.app, params)
         expected = 'failed to create repository `%s`' % (repo_name,)
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.parametrize('parent_group, dirty_name, expected_name', [
         (None, 'foo bar x', 'foo-bar-x'),
         ('foo', '/foo//bar x', 'foo/bar-x'),
         ('foo-bar', 'foo-bar //bar x', 'foo-bar/bar-x'),
     ])
     def test_create_repo_with_extra_slashes_in_name(
             self, backend, parent_group, dirty_name, expected_name):
 
         if parent_group:
             gr = fixture.create_repo_group(parent_group)
             assert gr.group_name == parent_group
 
         id_, params = build_data(
             self.apikey, 'create_repo',
             repo_name=dirty_name,
             repo_type=backend.alias,
             owner=TEST_USER_ADMIN_LOGIN,)
         response = api_call(self.app, params)
         expected ={
            "msg": "Created new repository `{}`".format(expected_name),
            "task": None,
            "success": True
         }
         assert_ok(id_, expected, response.body)
 
         repo = RepoModel().get_by_repo_name(expected_name)
         assert repo is not None
 
         expected = {
             'msg': 'Created new repository `%s`' % (expected_name,),
             'success': True,
             'task': None,
         }
         assert_ok(id_, expected, given=response.body)
         fixture.destroy_repo(expected_name)
         if parent_group:
             fixture.destroy_repo_group(parent_group)

rhodecode/api/tests/test_create_repo_group.py

@@ -1,288 +1,288 b''
 
 # Copyright (C) 2010-2023 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 mock
 import pytest
 
 from rhodecode.model.meta import Session
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.user import UserModel
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_ok, assert_error, crash)
 from rhodecode.tests.fixture import Fixture
 
 
 fixture = Fixture()
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestCreateRepoGroup(object):
     def test_api_create_repo_group(self):
         repo_group_name = 'api-repo-group'
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is None
 
         id_, params = build_data(
             self.apikey, 'create_repo_group',
             group_name=repo_group_name,
             owner=TEST_USER_ADMIN_LOGIN,)
         response = api_call(self.app, params)
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is not None
         ret = {
             'msg': 'Created new repo group `%s`' % (repo_group_name,),
             'repo_group': repo_group.get_api_data()
         }
         expected = ret
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo_group(repo_group_name)
 
     def test_api_create_repo_group_in_another_group(self):
         repo_group_name = 'api-repo-group'
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is None
         # create the parent
         fixture.create_repo_group(repo_group_name)
 
         full_repo_group_name = repo_group_name+'/'+repo_group_name
         id_, params = build_data(
             self.apikey, 'create_repo_group',
             group_name=full_repo_group_name,
             owner=TEST_USER_ADMIN_LOGIN,
             copy_permissions=True)
         response = api_call(self.app, params)
 
         repo_group = RepoGroupModel.cls.get_by_group_name(full_repo_group_name)
         assert repo_group is not None
         ret = {
             'msg': 'Created new repo group `%s`' % (full_repo_group_name,),
             'repo_group': repo_group.get_api_data()
         }
         expected = ret
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo_group(full_repo_group_name)
             fixture.destroy_repo_group(repo_group_name)
 
     def test_api_create_repo_group_in_another_group_not_existing(self):
         repo_group_name = 'api-repo-group-no'
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is None
 
         full_repo_group_name = repo_group_name+'/'+repo_group_name
         id_, params = build_data(
             self.apikey, 'create_repo_group',
             group_name=full_repo_group_name,
             owner=TEST_USER_ADMIN_LOGIN,
             copy_permissions=True)
         response = api_call(self.app, params)
         expected = {
             'repo_group':
                 'Parent repository group `{}` does not exist'.format(
                     repo_group_name)}
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_repo_group_that_exists(self):
         repo_group_name = 'api-repo-group'
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is None
 
         fixture.create_repo_group(repo_group_name)
         id_, params = build_data(
             self.apikey, 'create_repo_group',
             group_name=repo_group_name,
             owner=TEST_USER_ADMIN_LOGIN,)
         response = api_call(self.app, params)
         expected = {
             'unique_repo_group_name':
                 'Repository group with name `{}` already exists'.format(
                     repo_group_name)}
         try:
             assert_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo_group(repo_group_name)
 
     def test_api_create_repo_group_regular_user_wit_root_location_perms(
             self, user_util):
         regular_user = user_util.create_user()
         regular_user_api_key = regular_user.api_key
 
         repo_group_name = 'api-repo-group-by-regular-user'
 
         usr = UserModel().get_by_username(regular_user.username)
         usr.inherit_default_permissions = False
         Session().add(usr)
 
         UserModel().grant_perm(
             regular_user.username, 'hg.repogroup.create.true')
         Session().commit()
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is None
 
         id_, params = build_data(
             regular_user_api_key, 'create_repo_group',
             group_name=repo_group_name)
         response = api_call(self.app, params)
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is not None
         expected = {
             'msg': 'Created new repo group `%s`' % (repo_group_name,),
             'repo_group': repo_group.get_api_data()
         }
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo_group(repo_group_name)
 
     def test_api_create_repo_group_regular_user_with_admin_perms_to_parent(
             self, user_util):
 
         repo_group_name = 'api-repo-group-parent'
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is None
         # create the parent
         fixture.create_repo_group(repo_group_name)
 
         # user perms
         regular_user = user_util.create_user()
         regular_user_api_key = regular_user.api_key
 
         usr = UserModel().get_by_username(regular_user.username)
         usr.inherit_default_permissions = False
         Session().add(usr)
 
         RepoGroupModel().grant_user_permission(
             repo_group_name, regular_user.username, 'group.admin')
         Session().commit()
 
         full_repo_group_name = repo_group_name + '/' + repo_group_name
         id_, params = build_data(
             regular_user_api_key, 'create_repo_group',
             group_name=full_repo_group_name)
         response = api_call(self.app, params)
 
         repo_group = RepoGroupModel.cls.get_by_group_name(full_repo_group_name)
         assert repo_group is not None
         expected = {
             'msg': 'Created new repo group `{}`'.format(full_repo_group_name),
             'repo_group': repo_group.get_api_data()
         }
         try:
             assert_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo_group(full_repo_group_name)
             fixture.destroy_repo_group(repo_group_name)
 
     def test_api_create_repo_group_regular_user_no_permission_to_create_to_root_level(self):
         repo_group_name = 'api-repo-group'
 
         id_, params = build_data(
             self.apikey_regular, 'create_repo_group',
             group_name=repo_group_name)
         response = api_call(self.app, params)
 
         expected = {
             'repo_group':
-                u'You do not have the permission to store '
-                u'repository groups in the root location.'}
+                'You do not have the permission to store '
+                'repository groups in the root location.'}
         assert_error(id_, expected, given=response.body)
 
     def test_api_create_repo_group_regular_user_no_parent_group_perms(self):
         repo_group_name = 'api-repo-group-regular-user'
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is None
         # create the parent
         fixture.create_repo_group(repo_group_name)
 
         full_repo_group_name = repo_group_name+'/'+repo_group_name
 
         id_, params = build_data(
             self.apikey_regular, 'create_repo_group',
             group_name=full_repo_group_name)
         response = api_call(self.app, params)
 
         expected = {
             'repo_group':
-                u"You do not have the permissions to store "
-                u"repository groups inside repository group `{}`".format(repo_group_name)}
+                "You do not have the permissions to store "
+                "repository groups inside repository group `{}`".format(repo_group_name)}
         try:
             assert_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo_group(repo_group_name)
 
     def test_api_create_repo_group_regular_user_no_permission_to_specify_owner(
             self):
         repo_group_name = 'api-repo-group'
 
         id_, params = build_data(
             self.apikey_regular, 'create_repo_group',
             group_name=repo_group_name,
             owner=TEST_USER_ADMIN_LOGIN,)
         response = api_call(self.app, params)
 
         expected = "Only RhodeCode super-admin can specify `owner` param"
         assert_error(id_, expected, given=response.body)
 
     @mock.patch.object(RepoGroupModel, 'create', crash)
     def test_api_create_repo_group_exception_occurred(self):
         repo_group_name = 'api-repo-group'
 
         repo_group = RepoGroupModel.cls.get_by_group_name(repo_group_name)
         assert repo_group is None
 
         id_, params = build_data(
             self.apikey, 'create_repo_group',
             group_name=repo_group_name,
             owner=TEST_USER_ADMIN_LOGIN,)
         response = api_call(self.app, params)
         expected = 'failed to create repo group `%s`' % (repo_group_name,)
         assert_error(id_, expected, given=response.body)
 
     def test_create_group_with_extra_slashes_in_name(self, user_util):
         existing_repo_group = user_util.create_repo_group()
         dirty_group_name = '//{}//group2//'.format(
             existing_repo_group.group_name)
         cleaned_group_name = '{}/group2'.format(
             existing_repo_group.group_name)
 
         id_, params = build_data(
             self.apikey, 'create_repo_group',
             group_name=dirty_group_name,
             owner=TEST_USER_ADMIN_LOGIN,)
         response = api_call(self.app, params)
         repo_group = RepoGroupModel.cls.get_by_group_name(cleaned_group_name)
         expected = {
             'msg': 'Created new repo group `%s`' % (cleaned_group_name,),
             'repo_group': repo_group.get_api_data()
         }
         assert_ok(id_, expected, given=response.body)
         fixture.destroy_repo_group(cleaned_group_name)

rhodecode/api/tests/test_get_gist.py

@@ -1,101 +1,101 b''
 
 # Copyright (C) 2010-2023 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.lib.str_utils import safe_bytes
 from rhodecode.model.db import Gist
 from rhodecode.api.tests.utils import (
     build_data, api_call, assert_error, assert_ok)
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestApiGetGist(object):
     def test_api_get_gist(self, gist_util, http_host_only_stub):
         gist = gist_util.create_gist()
         gist_id = gist.gist_access_id
         gist_created_on = gist.created_on
         gist_modified_at = gist.modified_at
         id_, params = build_data(
             self.apikey, 'get_gist', gistid=gist_id, )
         response = api_call(self.app, params)
 
         expected = {
             'access_id': gist_id,
             'created_on': gist_created_on,
             'modified_at': gist_modified_at,
             'description': 'new-gist',
             'expires': -1.0,
             'gist_id': int(gist_id),
             'type': 'public',
             'url': 'http://%s/_admin/gists/%s' % (http_host_only_stub, gist_id,),
             'acl_level': Gist.ACL_LEVEL_PUBLIC,
             'content': None,
         }
 
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_gist_with_content(self, gist_util, http_host_only_stub):
         mapping = {
             b'filename1.txt': {'content': b'hello world'},
             safe_bytes('filename1ą.txt'): {'content': safe_bytes('hello worldę')}
         }
         gist = gist_util.create_gist(gist_mapping=mapping)
         gist_id = gist.gist_access_id
         gist_created_on = gist.created_on
         gist_modified_at = gist.modified_at
         id_, params = build_data(
             self.apikey, 'get_gist', gistid=gist_id, content=True)
         response = api_call(self.app, params)
 
         expected = {
             'access_id': gist_id,
             'created_on': gist_created_on,
             'modified_at': gist_modified_at,
             'description': 'new-gist',
             'expires': -1.0,
             'gist_id': int(gist_id),
             'type': 'public',
             'url': 'http://%s/_admin/gists/%s' % (http_host_only_stub, gist_id,),
             'acl_level': Gist.ACL_LEVEL_PUBLIC,
             'content': {
-                u'filename1.txt': u'hello world',
-                u'filename1ą.txt': u'hello worldę'
+                'filename1.txt': 'hello world',
+                'filename1ą.txt': 'hello worldę'
             },
         }
 
         assert_ok(id_, expected, given=response.body)
 
     def test_api_get_gist_not_existing(self):
         id_, params = build_data(
             self.apikey_regular, 'get_gist', gistid='12345', )
         response = api_call(self.app, params)
         expected = 'gist `%s` does not exist' % ('12345',)
         assert_error(id_, expected, given=response.body)
 
     def test_api_get_gist_private_gist_without_permission(self, gist_util):
         gist = gist_util.create_gist()
         gist_id = gist.gist_access_id
         id_, params = build_data(
             self.apikey_regular, 'get_gist', gistid=gist_id, )
         response = api_call(self.app, params)
 
         expected = 'gist `%s` does not exist' % (gist_id,)
         assert_error(id_, expected, given=response.body)

rhodecode/apps/admin/tests/test_admin_repos.py

@@ -1,497 +1,497 b''
 
 # Copyright (C) 2010-2023 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.request
 import urllib.parse
 import urllib.error
 
 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 repo_on_filesystem
 from rhodecode.tests.routes import route_path
 
 fixture = Fixture()
 
 
 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, xhr_header):
         repo = user_util.create_repo()
         repo_name = repo.repo_name
         response = self.app.get(
             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('[name=repo_type]')
         assert element.get('value') == 'git'
 
     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 ['hg', 'git', 'svn'] == [x.get('value') for x in assert_response.get_elements('[name=repo_type]')]
 
     @pytest.mark.parametrize(
         "suffix", ['', '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
 
         description_unicode = 'description for newly created repo' + suffix
         description = description_unicode
 
         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", ['', '_ąćę'], ids=['', 'non-ascii'])
     def test_create_in_group(
             self, autologin_user, backend, suffix, csrf_token):
         # create GROUP
         group_name = f'sometest_{backend.alias}'
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
 
         repo_name = f'ingroup{suffix}'
         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=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.")
+            "You do not have the permission to store repositories in "
+            "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):
         url_quoted_repo_name = urllib.parse.quote(repo_name)
 
         # run the check page that triggers the flash message
         response = self.app.get(
             route_path('repo_creating_check', repo_name=repo_name))
         assert response.json == {'result': True}
 
         flash_msg = 'Created repository <a href="/{}">{}</a>'.format(url_quoted_repo_name, 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=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(backend.alias)
 
         assert repo_on_filesystem(repo_name)

rhodecode/apps/login/tests/test_login.py

@@ -1,593 +1,593 b''
 # Copyright (C) 2010-2023 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.parse
 
 import mock
 import pytest
 
 
 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
 
 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.tests.routes import route_path
 
 fixture = Fixture()
 
 whitelist_view = ['RepoCommitsView:repo_commit_raw']
 
 
 @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('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('logout')
 
     def test_login_with_primary_email(self):
         user_email = '[e-mail]'
         response = self.app.post(route_path('login'),
                                  {'username': user_email,
                                   'password': 'test12'}, status=302)
         response = response.follow()
         session = response.get_session_from_response()
         user = session['rhodecode_user']
         assert user['username'] == user_email.split('@')[0]
         assert user['is_authenticated']
         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(self.app._pyramid_registry,
                                       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(self.app._pyramid_registry,
                                        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 = urllib.parse.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',
     ], 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'}, status=302)
         assert response.status == '302 Found'
         response = response.follow()
         assert response.status == '200 OK'
         assert response.request.path == '/'
 
     @pytest.mark.xfail(reason="newline params changed behaviour in python3")
     @pytest.mark.parametrize("url_came_from", [
         '/\r\nX-Forwarded-Host: \rhttp://example.org',
     ], ids=no_newline_id_generator)
     def test_login_bad_came_froms_404(self, url_came_from):
         _url = '{}?came_from={}'.format(route_path('login'), url_came_from)
         response = self.app.post(
             _url, {'username': 'test_admin', 'password': 'test12'}, status=302)
 
         response = response.follow()
         assert response.status == '404 Not Found'
 
     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': '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})
         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('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'
+        msg = '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'
+        msg = '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 = route_path('reset_password_confirmation', params={'key': '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):
                 # webtest uses linter to check if response is bytes,
                 # and we use memoryview here as a wrapper, quick turn-off
                 self.app.lint = 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):
                 # webtest uses linter to check if response is bytes,
                 # and we use memoryview here as a wrapper, quick turn-off
                 self.app.lint = 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):
                 # webtest uses linter to check if response is bytes,
                 # and we use memoryview here as a wrapper, quick turn-off
                 self.app.lint = 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):
                 # webtest uses linter to check if response is bytes,
                 # and we use memoryview here as a wrapper, quick turn-off
                 self.app.lint = 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):
                 # webtest uses linter to check if response is bytes,
                 # and we use memoryview here as a wrapper, quick turn-off
                 self.app.lint = 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/auth_external_test.py

@@ -1,108 +1,108 b''
 
 # Copyright (C) 2010-2023 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/
 
 """
 External module for testing plugins
 
 rhodecode.tests.auth_external_test
 
 """
 import logging
 import traceback
 
 from rhodecode.authentication.base import (
     RhodeCodeExternalAuthPlugin, hybrid_property)
 from rhodecode.model.db import User
 from rhodecode.lib.ext_json import formatted_json
 
 log = logging.getLogger(__name__)
 
 
 class RhodeCodeAuthPlugin(RhodeCodeExternalAuthPlugin):
     def __init__(self):
         self._logger = logging.getLogger(__name__)
 
     @hybrid_property
     def allows_creating_users(self):
         return True
 
     @hybrid_property
     def name(self):
-        return u"external_test"
+        return "external_test"
 
     def settings(self):
         settings = [
         ]
         return settings
 
     def use_fake_password(self):
         return True
 
     def user_activation_state(self):
         def_user_perms = User.get_default_user().AuthUser().permissions['global']
         return 'hg.extern_activate.auto' in def_user_perms
 
     def auth(self, userobj, username, password, settings, **kwargs):
         """
         Given a user object (which may be null), username, a plaintext password,
         and a settings object (containing all the keys needed as listed in settings()),
         authenticate this user's login attempt.
 
         Return None on failure. On success, return a dictionary of the form:
 
             see: RhodeCodeAuthPluginBase.auth_func_attrs
         This is later validated for correctness
         """
 
         if not username or not password:
             log.debug('Empty username or password skipping...')
             return None
 
         try:
             user_dn = username
 
             # # old attrs fetched from RhodeCode database
             admin = getattr(userobj, 'admin', False)
             active = getattr(userobj, 'active', True)
             email = getattr(userobj, 'email', '')
             firstname = getattr(userobj, 'firstname', '')
             lastname = getattr(userobj, 'lastname', '')
             extern_type = getattr(userobj, 'extern_type', '')
             #
             user_attrs = {
                 'username': username,
                 'firstname': firstname,
                 'lastname': lastname,
                 'groups': [],
                 'email': '[e-mail]' % username,
                 'admin': admin,
                 'active': active,
                 "active_from_extern": None,
                 'extern_name': user_dn,
                 'extern_type': extern_type,
             }
 
             log.debug('EXTERNAL user: \n%s', formatted_json(user_attrs))
             log.info('user `%s` authenticated correctly', user_attrs['username'])
 
             return user_attrs
 
         except (Exception,):
             log.error(traceback.format_exc())
             return None

rhodecode/tests/lib/auth_modules/test_auth_modules.py

@@ -1,188 +1,188 b''
 
 # Copyright (C) 2010-2023 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 mock
 import pytest
 
 from rhodecode.lib.auth import _RhodeCodeCryptoBCrypt
 from rhodecode.authentication.base import RhodeCodeAuthPluginBase
 from rhodecode.authentication.plugins.auth_ldap import RhodeCodeAuthPlugin
 from rhodecode.model import db
 
 
 class RcTestAuthPlugin(RhodeCodeAuthPluginBase):
 
     def name(self):
-        return u'stub_auth'
+        return 'stub_auth'
 
 
 def test_authenticate_returns_from_auth(stub_auth_data):
     plugin = RcTestAuthPlugin('stub_id')
     with mock.patch.object(plugin, 'auth') as auth_mock:
         auth_mock.return_value = stub_auth_data
         result = plugin._authenticate(mock.Mock(), 'test', 'password', {})
     assert stub_auth_data == result
 
 
 def test_authenticate_returns_empty_auth_data():
     auth_data = {}
     plugin = RcTestAuthPlugin('stub_id')
     with mock.patch.object(plugin, 'auth') as auth_mock:
         auth_mock.return_value = auth_data
         result = plugin._authenticate(mock.Mock(), 'test', 'password', {})
     assert auth_data == result
 
 
 def test_authenticate_skips_hash_migration_if_mismatch(stub_auth_data):
     stub_auth_data['_hash_migrate'] = 'new-hash'
     plugin = RcTestAuthPlugin('stub_id')
     with mock.patch.object(plugin, 'auth') as auth_mock:
         auth_mock.return_value = stub_auth_data
         result = plugin._authenticate(mock.Mock(), 'test', 'password', {})
 
     user = db.User.get_by_username(stub_auth_data['username'])
     assert user.password != 'new-hash'
     assert result == stub_auth_data
 
 
 def test_authenticate_migrates_to_new_hash(stub_auth_data):
     new_password = b'new-password'
     new_hash = _RhodeCodeCryptoBCrypt().hash_create(new_password)
     stub_auth_data['_hash_migrate'] = new_hash
     plugin = RcTestAuthPlugin('stub_id')
     with mock.patch.object(plugin, 'auth') as auth_mock:
         auth_mock.return_value = stub_auth_data
         result = plugin._authenticate(
             mock.Mock(), stub_auth_data['username'], new_password, {})
 
     user = db.User.get_by_username(stub_auth_data['username'])
     assert user.password == new_hash
     assert result == stub_auth_data
 
 
 @pytest.fixture()
 def stub_auth_data(user_util):
     user = user_util.create_user()
     data = {
         'username': user.username,
         'password': 'password',
         'email': '[e-mail]',
         'firstname': 'John',
         'lastname': 'Smith',
         'groups': [],
         'active': True,
         'admin': False,
         'extern_name': 'test',
         'extern_type': 'ldap',
         'active_from_extern': True
     }
     return data
 
 
 class TestRhodeCodeAuthPlugin(object):
     def setup_method(self, method):
         self.finalizers = []
         self.user = mock.Mock()
         self.user.username = 'test'
         self.user.password = 'old-password'
         self.fake_auth = {
             'username': 'test',
             'password': 'test',
             'email': '[e-mail]',
             'firstname': 'John',
             'lastname': 'Smith',
             'groups': [],
             'active': True,
             'admin': False,
             'extern_name': 'test',
             'extern_type': 'ldap',
             'active_from_extern': True
         }
 
     def teardown_method(self, method):
         if self.finalizers:
             for finalizer in self.finalizers:
                 finalizer()
             self.finalizers = []
 
     def test_fake_password_is_created_for_the_new_user(self):
         self._patch()
         auth_plugin = RhodeCodeAuthPlugin('stub_id')
         auth_plugin._authenticate(self.user, 'test', 'test', [])
         self.password_generator_mock.assert_called_once_with(length=16)
         create_user_kwargs = self.create_user_mock.call_args[1]
         assert create_user_kwargs['password'] == 'new-password'
 
     def test_fake_password_is_not_created_for_the_existing_user(self):
         self._patch()
         self.get_user_mock.return_value = self.user
         auth_plugin = RhodeCodeAuthPlugin('stub_id')
         auth_plugin._authenticate(self.user, 'test', 'test', [])
         assert self.password_generator_mock.called is False
         create_user_kwargs = self.create_user_mock.call_args[1]
         assert create_user_kwargs['password'] == self.user.password
 
     def _patch(self):
         get_user_patch = mock.patch('rhodecode.model.db.User.get_by_username')
         self.get_user_mock = get_user_patch.start()
         self.get_user_mock.return_value = None
         self.finalizers.append(get_user_patch.stop)
 
         create_user_patch = mock.patch(
             'rhodecode.model.user.UserModel.create_or_update')
         self.create_user_mock = create_user_patch.start()
         self.create_user_mock.return_value = None
         self.finalizers.append(create_user_patch.stop)
 
         auth_patch = mock.patch.object(RhodeCodeAuthPlugin, 'auth')
         self.auth_mock = auth_patch.start()
         self.auth_mock.return_value = self.fake_auth
         self.finalizers.append(auth_patch.stop)
 
         password_generator_patch = mock.patch(
             'rhodecode.lib.auth.PasswordGenerator.gen_password')
         self.password_generator_mock = password_generator_patch.start()
         self.password_generator_mock.return_value = 'new-password'
         self.finalizers.append(password_generator_patch.stop)
 
 
 def test_missing_ldap():
     from rhodecode.model.validators import Missing
 
     try:
         import ldap_not_existing
     except ImportError:
         # means that python-ldap is not installed
         ldap_not_existing = Missing
 
     # missing is singleton
     assert ldap_not_existing == Missing
 
 
 def test_import_ldap():
     from rhodecode.model.validators import Missing
 
     try:
         import ldap
     except ImportError:
         # means that python-ldap is not installed
         ldap = Missing
 
     # missing is singleton
     assert False is (ldap == Missing)

rhodecode/tests/vcs/test_git.py

@@ -1,1307 +1,1307 b''
 
 # Copyright (C) 2010-2023 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 datetime
 import mock
 import os
 import sys
 import shutil
 
 import pytest
 
 from rhodecode.lib.utils import make_db_config
 from rhodecode.lib.vcs.backends.base import Reference
 from rhodecode.lib.vcs.backends.git import (
     GitRepository, GitCommit, discover_git_version)
 from rhodecode.lib.vcs.exceptions import (
     RepositoryError, VCSError, NodeDoesNotExistError)
 from rhodecode.lib.vcs.nodes import (
     NodeKind, FileNode, DirNode, NodeState, SubModuleNode)
 from rhodecode.tests import TEST_GIT_REPO, TEST_GIT_REPO_CLONE, get_new_dir
 from rhodecode.tests.vcs.conftest import BackendTestMixin
 
 
 pytestmark = pytest.mark.backends("git")
 
 
 DIFF_FROM_REMOTE = br"""diff --git a/foobar b/foobar
 new file mode 100644
 index 0000000..f6ea049
 --- /dev/null
 +++ b/foobar
 @@ -0,0 +1 @@
 +foobar
 \ No newline at end of file
 diff --git a/foobar2 b/foobar2
 new file mode 100644
 index 0000000..e8c9d6b
 --- /dev/null
 +++ b/foobar2
 @@ -0,0 +1 @@
 +foobar2
 \ No newline at end of file
 """
 
 
 def callable_get_diff(*args, **kwargs):
     return DIFF_FROM_REMOTE
 
 
 class TestGitRepository(object):
 
     @pytest.fixture(autouse=True)
     def prepare(self, request, baseapp):
         self.repo = GitRepository(TEST_GIT_REPO, bare=True)
         self.repo.count()
 
     def get_clone_repo(self, tmpdir):
         """
         Return a non bare clone of the base repo.
         """
         clone_path = str(tmpdir.join('clone-repo'))
         repo_clone = GitRepository(
             clone_path, create=True, src_url=self.repo.path, bare=False)
 
         return repo_clone
 
     def get_empty_repo(self, tmpdir, bare=False):
         """
         Return a non bare empty repo.
         """
         clone_path = str(tmpdir.join('empty-repo'))
         return GitRepository(clone_path, create=True, bare=bare)
 
     def test_wrong_repo_path(self):
         wrong_repo_path = '/tmp/errorrepo_git'
         with pytest.raises(RepositoryError):
             GitRepository(wrong_repo_path)
 
     def test_repo_clone(self, tmp_path_factory):
         repo = GitRepository(TEST_GIT_REPO)
         clone_path = '{}_{}'.format(tmp_path_factory.mktemp('_'), TEST_GIT_REPO_CLONE)
         repo_clone = GitRepository(
             clone_path,
             src_url=TEST_GIT_REPO, create=True, do_workspace_checkout=True)
 
         assert len(repo.commit_ids) == len(repo_clone.commit_ids)
         # Checking hashes of commits should be enough
         for commit in repo.get_commits():
             raw_id = commit.raw_id
             assert raw_id == repo_clone.get_commit(raw_id).raw_id
 
     def test_repo_clone_without_create(self):
         with pytest.raises(RepositoryError):
             GitRepository(
                 TEST_GIT_REPO_CLONE + '_wo_create', src_url=TEST_GIT_REPO)
 
     def test_repo_clone_with_update(self, tmp_path_factory):
         repo = GitRepository(TEST_GIT_REPO)
         clone_path = '{}_{}_update'.format(tmp_path_factory.mktemp('_'), TEST_GIT_REPO_CLONE)
 
         repo_clone = GitRepository(
             clone_path,
             create=True, src_url=TEST_GIT_REPO, do_workspace_checkout=True)
         assert len(repo.commit_ids) == len(repo_clone.commit_ids)
 
         # check if current workdir was updated
         fpath = os.path.join(clone_path, 'MANIFEST.in')
         assert os.path.isfile(fpath)
 
     def test_repo_clone_without_update(self, tmp_path_factory):
         repo = GitRepository(TEST_GIT_REPO)
         clone_path = '{}_{}_without_update'.format(tmp_path_factory.mktemp('_'), TEST_GIT_REPO_CLONE)
         repo_clone = GitRepository(
             clone_path,
             create=True, src_url=TEST_GIT_REPO, do_workspace_checkout=False)
         assert len(repo.commit_ids) == len(repo_clone.commit_ids)
         # check if current workdir was *NOT* updated
         fpath = os.path.join(clone_path, 'MANIFEST.in')
         # Make sure it's not bare repo
         assert not repo_clone.bare
         assert not os.path.isfile(fpath)
 
     def test_repo_clone_into_bare_repo(self, tmp_path_factory):
         repo = GitRepository(TEST_GIT_REPO)
         clone_path = '{}_{}_bare.git'.format(tmp_path_factory.mktemp('_'), TEST_GIT_REPO_CLONE)
         repo_clone = GitRepository(
             clone_path, create=True, src_url=repo.path, bare=True)
         assert repo_clone.bare
 
     def test_create_repo_is_not_bare_by_default(self):
         repo = GitRepository(get_new_dir('not-bare-by-default'), create=True)
         assert not repo.bare
 
     def test_create_bare_repo(self):
         repo = GitRepository(get_new_dir('bare-repo'), create=True, bare=True)
         assert repo.bare
 
     def test_update_server_info(self):
         self.repo._update_server_info()
 
     def test_fetch(self, vcsbackend_git):
         # Note: This is a git specific part of the API, it's only implemented
         # by the git backend.
         source_repo = vcsbackend_git.repo
         target_repo = vcsbackend_git.create_repo(bare=True)
         target_repo.fetch(source_repo.path)
         # Note: Get a fresh instance, avoids caching trouble
         target_repo = vcsbackend_git.backend(target_repo.path)
         assert len(source_repo.commit_ids) == len(target_repo.commit_ids)
 
     def test_commit_ids(self):
         # there are 112 commits (by now)
         # so we can assume they would be available from now on
         subset = {'c1214f7e79e02fc37156ff215cd71275450cffc3',
                   '38b5fe81f109cb111f549bfe9bb6b267e10bc557',
                   'fa6600f6848800641328adbf7811fd2372c02ab2',
                   '102607b09cdd60e2793929c4f90478be29f85a17',
                   '49d3fd156b6f7db46313fac355dca1a0b94a0017',
                   '2d1028c054665b962fa3d307adfc923ddd528038',
                   'd7e0d30fbcae12c90680eb095a4f5f02505ce501',
                   'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
                   'dd80b0f6cf5052f17cc738c2951c4f2070200d7f',
                   '8430a588b43b5d6da365400117c89400326e7992',
                   'd955cd312c17b02143c04fa1099a352b04368118',
                   'f67b87e5c629c2ee0ba58f85197e423ff28d735b',
                   'add63e382e4aabc9e1afdc4bdc24506c269b7618',
                   'f298fe1189f1b69779a4423f40b48edf92a703fc',
                   'bd9b619eb41994cac43d67cf4ccc8399c1125808',
                   '6e125e7c890379446e98980d8ed60fba87d0f6d1',
                   'd4a54db9f745dfeba6933bf5b1e79e15d0af20bd',
                   '0b05e4ed56c802098dfc813cbe779b2f49e92500',
                   '191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
                   '45223f8f114c64bf4d6f853e3c35a369a6305520',
                   'ca1eb7957a54bce53b12d1a51b13452f95bc7c7e',
                   'f5ea29fc42ef67a2a5a7aecff10e1566699acd68',
                   '27d48942240f5b91dfda77accd2caac94708cc7d',
                   '622f0eb0bafd619d2560c26f80f09e3b0b0d78af',
                   'e686b958768ee96af8029fe19c6050b1a8dd3b2b'}
         assert subset.issubset(set(self.repo.commit_ids))
 
     def test_slicing(self):
         # 4 1 5 10 95
         for sfrom, sto, size in [(0, 4, 4), (1, 2, 1), (10, 15, 5),
                                  (10, 20, 10), (5, 100, 95)]:
             commit_ids = list(self.repo[sfrom:sto])
             assert len(commit_ids) == size
             assert commit_ids[0] == self.repo.get_commit(commit_idx=sfrom)
             assert commit_ids[-1] == self.repo.get_commit(commit_idx=sto - 1)
 
     def test_branches(self):
         # TODO: Need more tests here
         # Removed (those are 'remotes' branches for cloned repo)
         # assert 'master' in self.repo.branches
         # assert 'gittree' in self.repo.branches
         # assert 'web-branch' in self.repo.branches
         for __, commit_id in self.repo.branches.items():
             assert isinstance(self.repo.get_commit(commit_id), GitCommit)
 
     def test_tags(self):
         # TODO: Need more tests here
         assert 'v0.1.1' in self.repo.tags
         assert 'v0.1.2' in self.repo.tags
         for __, commit_id in self.repo.tags.items():
             assert isinstance(self.repo.get_commit(commit_id), GitCommit)
 
     def _test_single_commit_cache(self, commit_id):
         commit = self.repo.get_commit(commit_id)
         assert commit_id in self.repo.commits
         assert commit is self.repo.commits[commit_id]
 
     def test_initial_commit(self):
         commit_id = self.repo.commit_ids[0]
         init_commit = self.repo.get_commit(commit_id)
         init_author = init_commit.author
 
         assert init_commit.message == 'initial import\n'
         assert init_author == 'Marcin Kuzminski <[e-mail]>'
         assert init_author == init_commit.committer
         for path in ('vcs/__init__.py',
                      'vcs/backends/BaseRepository.py',
                      'vcs/backends/__init__.py'):
             assert isinstance(init_commit.get_node(path), FileNode)
         for path in ('', 'vcs', 'vcs/backends'):
             assert isinstance(init_commit.get_node(path), DirNode)
 
         with pytest.raises(NodeDoesNotExistError):
             init_commit.get_node(path='foobar')
 
         node = init_commit.get_node('vcs/')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.DIR
 
         node = init_commit.get_node('vcs')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.DIR
 
         node = init_commit.get_node('vcs/__init__.py')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.FILE
 
     def test_not_existing_commit(self):
         with pytest.raises(RepositoryError):
             self.repo.get_commit('f' * 40)
 
     def test_commit10(self):
 
         commit10 = self.repo.get_commit(self.repo.commit_ids[9])
         README = """===
 VCS
 ===
 
 Various Version Control System management abstraction layer for Python.
 
 Introduction
 ------------
 
 TODO: To be written...
 
 """
         node = commit10.get_node('README.rst')
         assert node.kind == NodeKind.FILE
         assert node.str_content == README
 
     def test_head(self):
         assert self.repo.head == self.repo.get_commit().raw_id
 
     def test_checkout_with_create(self, tmpdir):
         repo_clone = self.get_clone_repo(tmpdir)
 
         new_branch = 'new_branch'
         assert repo_clone._current_branch() == 'master'
         assert set(repo_clone.branches) == {'master'}
         repo_clone._checkout(new_branch, create=True)
 
         # Branches is a lazy property so we need to recrete the Repo object.
         repo_clone = GitRepository(repo_clone.path)
         assert set(repo_clone.branches) == {'master', new_branch}
         assert repo_clone._current_branch() == new_branch
 
     def test_checkout(self, tmpdir):
         repo_clone = self.get_clone_repo(tmpdir)
 
         repo_clone._checkout('new_branch', create=True)
         repo_clone._checkout('master')
 
         assert repo_clone._current_branch() == 'master'
 
     def test_checkout_same_branch(self, tmpdir):
         repo_clone = self.get_clone_repo(tmpdir)
 
         repo_clone._checkout('master')
         assert repo_clone._current_branch() == 'master'
 
     def test_checkout_branch_already_exists(self, tmpdir):
         repo_clone = self.get_clone_repo(tmpdir)
 
         with pytest.raises(RepositoryError):
             repo_clone._checkout('master', create=True)
 
     def test_checkout_bare_repo(self):
         with pytest.raises(RepositoryError):
             self.repo._checkout('master')
 
     def test_current_branch_bare_repo(self):
         with pytest.raises(RepositoryError):
             self.repo._current_branch()
 
     def test_current_branch_empty_repo(self, tmpdir):
         repo = self.get_empty_repo(tmpdir)
         assert repo._current_branch() is None
 
     def test_local_clone(self, tmp_path_factory):
         clone_path = str(tmp_path_factory.mktemp('test-local-clone'))
         self.repo._local_clone(clone_path, 'master')
         repo_clone = GitRepository(clone_path)
 
         assert self.repo.commit_ids == repo_clone.commit_ids
 
     def test_local_clone_with_specific_branch(self, tmpdir):
         source_repo = self.get_clone_repo(tmpdir)
 
         # Create a new branch in source repo
         new_branch_commit = source_repo.commit_ids[-3]
         source_repo._checkout(new_branch_commit)
         source_repo._checkout('new_branch', create=True)
 
         clone_path = str(tmpdir.join('git-clone-path-1'))
         source_repo._local_clone(clone_path, 'new_branch')
         repo_clone = GitRepository(clone_path)
 
         assert source_repo.commit_ids[:-3 + 1] == repo_clone.commit_ids
 
         clone_path = str(tmpdir.join('git-clone-path-2'))
         source_repo._local_clone(clone_path, 'master')
         repo_clone = GitRepository(clone_path)
 
         assert source_repo.commit_ids == repo_clone.commit_ids
 
     def test_local_clone_fails_if_target_exists(self):
         with pytest.raises(RepositoryError):
             self.repo._local_clone(self.repo.path, 'master')
 
     def test_local_fetch(self, tmpdir):
         target_repo = self.get_empty_repo(tmpdir)
         source_repo = self.get_clone_repo(tmpdir)
 
         # Create a new branch in source repo
         master_commit = source_repo.commit_ids[-1]
         new_branch_commit = source_repo.commit_ids[-3]
         source_repo._checkout(new_branch_commit)
         source_repo._checkout('new_branch', create=True)
 
         target_repo._local_fetch(source_repo.path, 'new_branch')
         assert target_repo._last_fetch_heads() == [new_branch_commit]
 
         target_repo._local_fetch(source_repo.path, 'master')
         assert target_repo._last_fetch_heads() == [master_commit]
 
     def test_local_fetch_from_bare_repo(self, tmpdir):
         target_repo = self.get_empty_repo(tmpdir)
         target_repo._local_fetch(self.repo.path, 'master')
 
         master_commit = self.repo.commit_ids[-1]
         assert target_repo._last_fetch_heads() == [master_commit]
 
     def test_local_fetch_from_same_repo(self):
         with pytest.raises(ValueError):
             self.repo._local_fetch(self.repo.path, 'master')
 
     def test_local_fetch_branch_does_not_exist(self, tmpdir):
         target_repo = self.get_empty_repo(tmpdir)
 
         with pytest.raises(RepositoryError):
             target_repo._local_fetch(self.repo.path, 'new_branch')
 
     def test_local_pull(self, tmpdir):
         target_repo = self.get_empty_repo(tmpdir)
         source_repo = self.get_clone_repo(tmpdir)
 
         # Create a new branch in source repo
         master_commit = source_repo.commit_ids[-1]
         new_branch_commit = source_repo.commit_ids[-3]
         source_repo._checkout(new_branch_commit)
         source_repo._checkout('new_branch', create=True)
 
         target_repo._local_pull(source_repo.path, 'new_branch')
         target_repo = GitRepository(target_repo.path)
         assert target_repo.head == new_branch_commit
 
         target_repo._local_pull(source_repo.path, 'master')
         target_repo = GitRepository(target_repo.path)
         assert target_repo.head == master_commit
 
     def test_local_pull_in_bare_repo(self):
         with pytest.raises(RepositoryError):
             self.repo._local_pull(self.repo.path, 'master')
 
     def test_local_merge(self, tmpdir):
         target_repo = self.get_empty_repo(tmpdir)
         source_repo = self.get_clone_repo(tmpdir)
 
         # Create a new branch in source repo
         master_commit = source_repo.commit_ids[-1]
         new_branch_commit = source_repo.commit_ids[-3]
         source_repo._checkout(new_branch_commit)
         source_repo._checkout('new_branch', create=True)
 
         # This is required as one cannot do a -ff-only merge in an empty repo.
         target_repo._local_pull(source_repo.path, 'new_branch')
 
         target_repo._local_fetch(source_repo.path, 'master')
         merge_message = 'Merge message\n\nDescription:...'
         user_name = 'Albert Einstein'
         user_email = '[e-mail]'
         target_repo._local_merge(merge_message, user_name, user_email,
                                  target_repo._last_fetch_heads())
 
         target_repo = GitRepository(target_repo.path)
         assert target_repo.commit_ids[-2] == master_commit
         last_commit = target_repo.get_commit(target_repo.head)
         assert last_commit.message.strip() == merge_message
         assert last_commit.author == '%s <%s>' % (user_name, user_email)
 
         assert not os.path.exists(
             os.path.join(target_repo.path, '.git', 'MERGE_HEAD'))
 
     def test_local_merge_raises_exception_on_conflict(self, vcsbackend_git):
         target_repo = vcsbackend_git.create_repo(number_of_commits=1)
         vcsbackend_git.ensure_file(b'README', b'I will conflict with you!!!')
 
         target_repo._local_fetch(self.repo.path, 'master')
         with pytest.raises(RepositoryError):
             target_repo._local_merge(
                 'merge_message', 'user name', '[e-mail]',
                 target_repo._last_fetch_heads())
 
         # Check we are not left in an intermediate merge state
         assert not os.path.exists(
             os.path.join(target_repo.path, '.git', 'MERGE_HEAD'))
 
     def test_local_merge_into_empty_repo(self, tmpdir):
         target_repo = self.get_empty_repo(tmpdir)
 
         # This is required as one cannot do a -ff-only merge in an empty repo.
         target_repo._local_fetch(self.repo.path, 'master')
         with pytest.raises(RepositoryError):
             target_repo._local_merge(
                 'merge_message', 'user name', '[e-mail]',
                 target_repo._last_fetch_heads())
 
     def test_local_merge_in_bare_repo(self):
         with pytest.raises(RepositoryError):
             self.repo._local_merge(
                 'merge_message', 'user name', '[e-mail]', None)
 
     def test_local_push_non_bare(self, tmpdir):
         target_repo = self.get_empty_repo(tmpdir)
 
         pushed_branch = 'pushed_branch'
         self.repo._local_push('master', target_repo.path, pushed_branch)
         # Fix the HEAD of the target repo, or otherwise GitRepository won't
         # report any branches.
         with open(os.path.join(target_repo.path, '.git', 'HEAD'), 'w') as f:
             f.write('ref: refs/heads/%s' % pushed_branch)
 
         target_repo = GitRepository(target_repo.path)
 
         assert (target_repo.branches[pushed_branch] ==
                 self.repo.branches['master'])
 
     def test_local_push_bare(self, tmpdir):
         target_repo = self.get_empty_repo(tmpdir, bare=True)
 
         pushed_branch = 'pushed_branch'
         self.repo._local_push('master', target_repo.path, pushed_branch)
         # Fix the HEAD of the target repo, or otherwise GitRepository won't
         # report any branches.
         with open(os.path.join(target_repo.path, 'HEAD'), 'w') as f:
             f.write('ref: refs/heads/%s' % pushed_branch)
 
         target_repo = GitRepository(target_repo.path)
 
         assert (target_repo.branches[pushed_branch] ==
                 self.repo.branches['master'])
 
     def test_local_push_non_bare_target_branch_is_checked_out(self, tmpdir):
         target_repo = self.get_clone_repo(tmpdir)
 
         pushed_branch = 'pushed_branch'
         # Create a new branch in source repo
         new_branch_commit = target_repo.commit_ids[-3]
         target_repo._checkout(new_branch_commit)
         target_repo._checkout(pushed_branch, create=True)
 
         self.repo._local_push('master', target_repo.path, pushed_branch)
 
         target_repo = GitRepository(target_repo.path)
 
         assert (target_repo.branches[pushed_branch] ==
                 self.repo.branches['master'])
 
     def test_local_push_raises_exception_on_conflict(self, vcsbackend_git):
         target_repo = vcsbackend_git.create_repo(number_of_commits=1)
         with pytest.raises(RepositoryError):
             self.repo._local_push('master', target_repo.path, 'master')
 
     def test_hooks_can_be_enabled_via_env_variable_for_local_push(self, tmpdir):
         target_repo = self.get_empty_repo(tmpdir, bare=True)
 
         with mock.patch.object(self.repo, 'run_git_command') as run_mock:
             self.repo._local_push(
                 'master', target_repo.path, 'master', enable_hooks=True)
         env = run_mock.call_args[1]['extra_env']
         assert 'RC_SKIP_HOOKS' not in env
 
     def _add_failing_hook(self, repo_path, hook_name, bare=False):
         path_components = (
             ['hooks', hook_name] if bare else ['.git', 'hooks', hook_name])
         hook_path = os.path.join(repo_path, *path_components)
         with open(hook_path, 'w') as f:
             script_lines = [
                 '#!%s' % sys.executable,
                 'import os',
                 'import sys',
                 'if os.environ.get("RC_SKIP_HOOKS"):',
                 '    sys.exit(0)',
                 'sys.exit(1)',
             ]
             f.write('\n'.join(script_lines))
         os.chmod(hook_path, 0o755)
 
     def test_local_push_does_not_execute_hook(self, tmpdir):
         target_repo = self.get_empty_repo(tmpdir)
 
         pushed_branch = 'pushed_branch'
         self._add_failing_hook(target_repo.path, 'pre-receive')
         self.repo._local_push('master', target_repo.path, pushed_branch)
         # Fix the HEAD of the target repo, or otherwise GitRepository won't
         # report any branches.
         with open(os.path.join(target_repo.path, '.git', 'HEAD'), 'w') as f:
             f.write('ref: refs/heads/%s' % pushed_branch)
 
         target_repo = GitRepository(target_repo.path)
 
         assert (target_repo.branches[pushed_branch] ==
                 self.repo.branches['master'])
 
     def test_local_push_executes_hook(self, tmpdir):
         target_repo = self.get_empty_repo(tmpdir, bare=True)
         self._add_failing_hook(target_repo.path, 'pre-receive', bare=True)
         with pytest.raises(RepositoryError):
             self.repo._local_push(
                 'master', target_repo.path, 'master', enable_hooks=True)
 
     def test_maybe_prepare_merge_workspace(self):
         workspace = self.repo._maybe_prepare_merge_workspace(
             2, 'pr2', Reference('branch', 'master', 'unused'),
             Reference('branch', 'master', 'unused'))
 
         assert os.path.isdir(workspace)
         workspace_repo = GitRepository(workspace)
         assert workspace_repo.branches == self.repo.branches
 
         # Calling it a second time should also succeed
         workspace = self.repo._maybe_prepare_merge_workspace(
             2, 'pr2', Reference('branch', 'master', 'unused'),
             Reference('branch', 'master', 'unused'))
         assert os.path.isdir(workspace)
 
     def test_maybe_prepare_merge_workspace_different_refs(self):
         workspace = self.repo._maybe_prepare_merge_workspace(
             2, 'pr2', Reference('branch', 'master', 'unused'),
             Reference('branch', 'develop', 'unused'))
 
         assert os.path.isdir(workspace)
         workspace_repo = GitRepository(workspace)
         assert workspace_repo.branches == self.repo.branches
 
         # Calling it a second time should also succeed
         workspace = self.repo._maybe_prepare_merge_workspace(
             2, 'pr2', Reference('branch', 'master', 'unused'),
             Reference('branch', 'develop', 'unused'))
         assert os.path.isdir(workspace)
 
     def test_cleanup_merge_workspace(self):
         workspace = self.repo._maybe_prepare_merge_workspace(
             2, 'pr3', Reference('branch', 'master', 'unused'),
             Reference('branch', 'master', 'unused'))
         self.repo.cleanup_merge_workspace(2, 'pr3')
 
         assert not os.path.exists(workspace)
 
     def test_cleanup_merge_workspace_invalid_workspace_id(self):
         # No assert: because in case of an inexistent workspace this function
         # should still succeed.
         self.repo.cleanup_merge_workspace(1, 'pr4')
 
     def test_set_refs(self):
         test_ref = 'refs/test-refs/abcde'
         test_commit_id = 'ecb86e1f424f2608262b130db174a7dfd25a6623'
 
         self.repo.set_refs(test_ref, test_commit_id)
         stdout, _ = self.repo.run_git_command(['show-ref'])
         assert test_ref in stdout
         assert test_commit_id in stdout
 
     def test_remove_ref(self):
         test_ref = 'refs/test-refs/abcde'
         test_commit_id = 'ecb86e1f424f2608262b130db174a7dfd25a6623'
         self.repo.set_refs(test_ref, test_commit_id)
         stdout, _ = self.repo.run_git_command(['show-ref'])
         assert test_ref in stdout
         assert test_commit_id in stdout
 
         self.repo.remove_ref(test_ref)
         stdout, _ = self.repo.run_git_command(['show-ref'])
         assert test_ref not in stdout
         assert test_commit_id not in stdout
 
 
 class TestGitCommit(object):
 
     @pytest.fixture(autouse=True)
     def prepare(self):
         self.repo = GitRepository(TEST_GIT_REPO)
 
     def test_default_commit(self):
         tip = self.repo.get_commit()
         assert tip == self.repo.get_commit(None)
         assert tip == self.repo.get_commit('tip')
 
     def test_root_node(self):
         tip = self.repo.get_commit()
         assert tip.root is tip.get_node('')
 
     def test_lazy_fetch(self):
         """
         Test if commit's nodes expands and are cached as we walk through
         the commit. This test is somewhat hard to write as order of tests
         is a key here. Written by running command after command in a shell.
         """
         commit_id = '2a13f185e4525f9d4b59882791a2d397b90d5ddc'
         assert commit_id in self.repo.commit_ids
         commit = self.repo.get_commit(commit_id)
         assert len(commit.nodes) == 0
         root = commit.root
         assert len(commit.nodes) == 1
         assert len(root.nodes) == 8
         # accessing root.nodes updates commit.nodes
         assert len(commit.nodes) == 9
 
         docs = root.get_node('docs')
         # we haven't yet accessed anything new as docs dir was already cached
         assert len(commit.nodes) == 9
         assert len(docs.nodes) == 8
         # accessing docs.nodes updates commit.nodes
         assert len(commit.nodes) == 17
 
         assert docs is commit.get_node('docs')
         assert docs is root.nodes[0]
         assert docs is root.dirs[0]
         assert docs is commit.get_node('docs')
 
     def test_nodes_with_commit(self):
         commit_id = '2a13f185e4525f9d4b59882791a2d397b90d5ddc'
         commit = self.repo.get_commit(commit_id)
         root = commit.root
         docs = root.get_node('docs')
         assert docs is commit.get_node('docs')
         api = docs.get_node('api')
         assert api is commit.get_node('docs/api')
         index = api.get_node('index.rst')
         assert index is commit.get_node('docs/api/index.rst')
         assert index is commit.get_node('docs')\
             .get_node('api')\
             .get_node('index.rst')
 
     def test_branch_and_tags(self):
         """
         rev0 = self.repo.commit_ids[0]
         commit0 = self.repo.get_commit(rev0)
         assert commit0.branch == 'master'
         assert commit0.tags == []
 
         rev10 = self.repo.commit_ids[10]
         commit10 = self.repo.get_commit(rev10)
         assert commit10.branch == 'master'
         assert commit10.tags == []
 
         rev44 = self.repo.commit_ids[44]
         commit44 = self.repo.get_commit(rev44)
         assert commit44.branch == 'web-branch'
 
         tip = self.repo.get_commit('tip')
         assert 'tip' in tip.tags
         """
         # Those tests would fail - branches are now going
         # to be changed at main API in order to support git backend
         pass
 
     def test_file_size(self):
         to_check = (
             ('c1214f7e79e02fc37156ff215cd71275450cffc3',
                 'vcs/backends/BaseRepository.py', 502),
             ('d7e0d30fbcae12c90680eb095a4f5f02505ce501',
                 'vcs/backends/hg.py', 854),
             ('6e125e7c890379446e98980d8ed60fba87d0f6d1',
                 'setup.py', 1068),
 
             ('d955cd312c17b02143c04fa1099a352b04368118',
                 'vcs/backends/base.py', 2921),
             ('ca1eb7957a54bce53b12d1a51b13452f95bc7c7e',
                 'vcs/backends/base.py', 3936),
             ('f50f42baeed5af6518ef4b0cb2f1423f3851a941',
                 'vcs/backends/base.py', 6189),
         )
         for commit_id, path, size in to_check:
             node = self.repo.get_commit(commit_id).get_node(path)
             assert node.is_file()
             assert node.size == size
 
     def test_file_history_from_commits(self):
         node = self.repo[10].get_node('setup.py')
         commit_ids = [commit.raw_id for commit in node.history]
         assert ['ff7ca51e58c505fec0dd2491de52c622bb7a806b'] == commit_ids
 
         node = self.repo[20].get_node('setup.py')
         node_ids = [commit.raw_id for commit in node.history]
         assert ['191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b'] == node_ids
 
         # special case we check history from commit that has this particular
         # file changed this means we check if it's included as well
         node = self.repo.get_commit('191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e') \
             .get_node('setup.py')
         node_ids = [commit.raw_id for commit in node.history]
         assert ['191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b'] == node_ids
 
     def test_file_history(self):
         # we can only check if those commits are present in the history
         # as we cannot update this test every time file is changed
         files = {
             'setup.py': [
                 '54386793436c938cff89326944d4c2702340037d',
                 '51d254f0ecf5df2ce50c0b115741f4cf13985dab',
                 '998ed409c795fec2012b1c0ca054d99888b22090',
                 '5e0eb4c47f56564395f76333f319d26c79e2fb09',
                 '0115510b70c7229dbc5dc49036b32e7d91d23acd',
                 '7cb3fd1b6d8c20ba89e2264f1c8baebc8a52d36e',
                 '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                 '191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             ],
             'vcs/nodes.py': [
                 '33fa3223355104431402a888fa77a4e9956feb3e',
                 'fa014c12c26d10ba682fadb78f2a11c24c8118e1',
                 'e686b958768ee96af8029fe19c6050b1a8dd3b2b',
                 'ab5721ca0a081f26bf43d9051e615af2cc99952f',
                 'c877b68d18e792a66b7f4c529ea02c8f80801542',
                 '4313566d2e417cb382948f8d9d7c765330356054',
                 '6c2303a793671e807d1cfc70134c9ca0767d98c2',
                 '54386793436c938cff89326944d4c2702340037d',
                 '54000345d2e78b03a99d561399e8e548de3f3203',
                 '1c6b3677b37ea064cb4b51714d8f7498f93f4b2b',
                 '2d03ca750a44440fb5ea8b751176d1f36f8e8f46',
                 '2a08b128c206db48c2f0b8f70df060e6db0ae4f8',
                 '30c26513ff1eb8e5ce0e1c6b477ee5dc50e2f34b',
                 'ac71e9503c2ca95542839af0ce7b64011b72ea7c',
                 '12669288fd13adba2a9b7dd5b870cc23ffab92d2',
                 '5a0c84f3e6fe3473e4c8427199d5a6fc71a9b382',
                 '12f2f5e2b38e6ff3fbdb5d722efed9aa72ecb0d5',
                 '5eab1222a7cd4bfcbabc218ca6d04276d4e27378',
                 'f50f42baeed5af6518ef4b0cb2f1423f3851a941',
                 'd7e390a45f6aa96f04f5e7f583ad4f867431aa25',
                 'f15c21f97864b4f071cddfbf2750ec2e23859414',
                 'e906ef056cf539a4e4e5fc8003eaf7cf14dd8ade',
                 'ea2b108b48aa8f8c9c4a941f66c1a03315ca1c3b',
                 '84dec09632a4458f79f50ddbbd155506c460b4f9',
                 '0115510b70c7229dbc5dc49036b32e7d91d23acd',
                 '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                 '3bf1c5868e570e39569d094f922d33ced2fa3b2b',
                 'b8d04012574729d2c29886e53b1a43ef16dd00a1',
                 '6970b057cffe4aab0a792aa634c89f4bebf01441',
                 'dd80b0f6cf5052f17cc738c2951c4f2070200d7f',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             ],
             'vcs/backends/git.py': [
                 '4cf116ad5a457530381135e2f4c453e68a1b0105',
                 '9a751d84d8e9408e736329767387f41b36935153',
                 'cb681fb539c3faaedbcdf5ca71ca413425c18f01',
                 '428f81bb652bcba8d631bce926e8834ff49bdcc6',
                 '180ab15aebf26f98f714d8c68715e0f05fa6e1c7',
                 '2b8e07312a2e89e92b90426ab97f349f4bce2a3a',
                 '50e08c506174d8645a4bb517dd122ac946a0f3bf',
                 '54000345d2e78b03a99d561399e8e548de3f3203',
             ],
         }
         for path, commit_ids in files.items():
             node = self.repo.get_commit(commit_ids[0]).get_node(path)
             node_ids = [commit.raw_id for commit in node.history]
             assert set(commit_ids).issubset(set(node_ids)), (
                 "We assumed that %s is subset of commit_ids for which file %s "
                 "has been changed, and history of that node returned: %s"
                 % (commit_ids, path, node_ids))
 
     def test_file_annotate(self):
         files = {
             'vcs/backends/__init__.py': {
                 'c1214f7e79e02fc37156ff215cd71275450cffc3': {
                     'lines_no': 1,
                     'commits': [
                         'c1214f7e79e02fc37156ff215cd71275450cffc3',
                     ],
                 },
                 '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647': {
                     'lines_no': 21,
                     'commits': [
                         '49d3fd156b6f7db46313fac355dca1a0b94a0017',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                     ],
                 },
                 'e29b67bd158580fc90fc5e9111240b90e6e86064': {
                     'lines_no': 32,
                     'commits': [
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '5eab1222a7cd4bfcbabc218ca6d04276d4e27378',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '54000345d2e78b03a99d561399e8e548de3f3203',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '78c3f0c23b7ee935ec276acb8b8212444c33c396',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '78c3f0c23b7ee935ec276acb8b8212444c33c396',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                     ],
                 },
             },
         }
 
         for fname, commit_dict in files.items():
             for commit_id, __ in commit_dict.items():
                 commit = self.repo.get_commit(commit_id)
 
                 l1_1 = [x[1] for x in commit.get_file_annotate(fname)]
                 l1_2 = [x[2]().raw_id for x in commit.get_file_annotate(fname)]
                 assert l1_1 == l1_2
                 l1 = l1_1
                 l2 = files[fname][commit_id]['commits']
                 assert l1 == l2, (
                     "The lists of commit_ids for %s@commit_id %s"
                     "from annotation list should match each other, "
                     "got \n%s \nvs \n%s " % (fname, commit_id, l1, l2))
 
     def test_files_state(self):
         """
         Tests state of FileNodes.
         """
         node = self.repo\
             .get_commit('e6ea6d16e2f26250124a1f4b4fe37a912f9d86a0')\
             .get_node('vcs/utils/diffs.py')
         assert node.state, NodeState.ADDED
         assert node.added
         assert not node.changed
         assert not node.not_changed
         assert not node.removed
 
         node = self.repo\
             .get_commit('33fa3223355104431402a888fa77a4e9956feb3e')\
             .get_node('.hgignore')
         assert node.state, NodeState.CHANGED
         assert not node.added
         assert node.changed
         assert not node.not_changed
         assert not node.removed
 
         node = self.repo\
             .get_commit('e29b67bd158580fc90fc5e9111240b90e6e86064')\
             .get_node('setup.py')
         assert node.state, NodeState.NOT_CHANGED
         assert not node.added
         assert not node.changed
         assert node.not_changed
         assert not node.removed
 
         # If node has REMOVED state then trying to fetch it would raise
         # CommitError exception
         commit = self.repo.get_commit(
             'fa6600f6848800641328adbf7811fd2372c02ab2')
         path = 'vcs/backends/BaseRepository.py'
         with pytest.raises(NodeDoesNotExistError):
             commit.get_node(path)
         # but it would be one of ``removed`` (commit's attribute)
         assert path in [rf.path for rf in commit.removed]
 
         commit = self.repo.get_commit(
             '54386793436c938cff89326944d4c2702340037d')
         changed = [
             'setup.py', 'tests/test_nodes.py', 'vcs/backends/hg.py',
             'vcs/nodes.py']
         assert set(changed) == set([f.path for f in commit.changed])
 
     def test_unicode_branch_refs(self):
         unicode_branches = {
             'refs/heads/unicode': '6c0ce52b229aa978889e91b38777f800e85f330b',
             u'refs/heads/uniçö∂e': 'ürl',
         }
         with mock.patch(
             ("rhodecode.lib.vcs.backends.git.repository"
                 ".GitRepository._refs"),
                 unicode_branches):
             branches = self.repo.branches
 
         assert 'unicode' in branches
-        assert u'uniçö∂e' in branches
+        assert 'uniçö∂e' in branches
 
     def test_unicode_tag_refs(self):
         unicode_tags = {
             'refs/tags/unicode': '6c0ce52b229aa978889e91b38777f800e85f330b',
             u'refs/tags/uniçö∂e': '6c0ce52b229aa978889e91b38777f800e85f330b',
         }
         with mock.patch(
             ("rhodecode.lib.vcs.backends.git.repository"
                 ".GitRepository._refs"),
                 unicode_tags):
             tags = self.repo.tags
 
         assert 'unicode' in tags
-        assert u'uniçö∂e' in tags
+        assert 'uniçö∂e' in tags
 
     def test_commit_message_is_unicode(self):
         for commit in self.repo:
             assert type(commit.message) == str
 
     def test_commit_author_is_unicode(self):
         for commit in self.repo:
             assert type(commit.author) == str
 
     def test_repo_files_content_types(self):
         commit = self.repo.get_commit()
         for node in commit.get_node('/'):
             if node.is_file():
                 assert type(node.content) == bytes
                 assert type(node.str_content) == str
 
     def test_wrong_path(self):
         # There is 'setup.py' in the root dir but not there:
         path = 'foo/bar/setup.py'
         tip = self.repo.get_commit()
         with pytest.raises(VCSError):
             tip.get_node(path)
 
     @pytest.mark.parametrize("author_email, commit_id", [
         ('[e-mail]', 'c1214f7e79e02fc37156ff215cd71275450cffc3'),
         ('[e-mail]',
          'ff7ca51e58c505fec0dd2491de52c622bb7a806b'),
         ('none@none', '8430a588b43b5d6da365400117c89400326e7992'),
     ])
     def test_author_email(self, author_email, commit_id):
         commit = self.repo.get_commit(commit_id)
         assert author_email == commit.author_email
 
     @pytest.mark.parametrize("author, commit_id", [
         ('Marcin Kuzminski', 'c1214f7e79e02fc37156ff215cd71275450cffc3'),
         ('Lukasz Balcerzak', 'ff7ca51e58c505fec0dd2491de52c622bb7a806b'),
         ('marcink', '8430a588b43b5d6da365400117c89400326e7992'),
     ])
     def test_author_username(self, author, commit_id):
         commit = self.repo.get_commit(commit_id)
         assert author == commit.author_name
 
 
 class TestLargeFileRepo(object):
 
     def test_large_file(self, backend_git):
         conf = make_db_config()
         repo = backend_git.create_test_repo('largefiles', conf)
 
         tip = repo.scm_instance().get_commit()
 
         # extract stored LF node into the origin cache
         lfs_store = os.path.join(repo.repo_path, repo.repo_name, 'lfs_store')
 
         oid = '7b331c02e313c7599d5a90212e17e6d3cb729bd2e1c9b873c302a63c95a2f9bf'
         oid_path = os.path.join(lfs_store, oid)
         # Todo: oid path depends on LFSOidStorage.store_suffix. Once it will be changed update below line accordingly
         oid_destination = os.path.join(
             conf.get('vcs_git_lfs', 'store_location'), f'objects/{oid[:2]}/{oid[2:4]}/{oid}')
 
         os.makedirs(os.path.dirname(oid_destination))
         shutil.copy(oid_path, oid_destination)
 
         node = tip.get_node('1MB.zip')
 
         lf_node = node.get_largefile_node()
 
         assert lf_node.is_largefile() is True
         assert lf_node.size == 1024000
         assert lf_node.name == '1MB.zip'
 
 
 @pytest.mark.usefixtures("vcs_repository_support")
 class TestGitSpecificWithRepo(BackendTestMixin):
 
     @classmethod
     def _get_commits(cls):
         return [
             {
                 'message': 'Initial',
                 'author': 'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode(b'foobar/static/js/admin/base.js', content=b'base'),
                     FileNode(b'foobar/static/admin', content=b'admin', mode=0o120000),  # this is a link
                     FileNode(b'foo', content=b'foo'),
                 ],
             },
             {
                 'message': 'Second',
                 'author': 'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'added': [
                     FileNode(b'foo2', content=b'foo2'),
                 ],
             },
         ]
 
     def test_paths_slow_traversing(self):
         commit = self.repo.get_commit()
         assert commit.get_node('foobar').get_node('static').get_node('js')\
             .get_node('admin').get_node('base.js').content == b'base'
 
     def test_paths_fast_traversing(self):
         commit = self.repo.get_commit()
         assert commit.get_node('foobar/static/js/admin/base.js').content == b'base'
 
     def test_get_diff_runs_git_command_with_hashes(self):
         comm1 = self.repo[0]
         comm2 = self.repo[1]
 
         with mock.patch.object(self.repo, '_remote', return_value=mock.Mock()) as remote_mock:
             remote_mock.diff = mock.MagicMock(side_effect=callable_get_diff)
             self.repo.get_diff(comm1, comm2)
 
         remote_mock.diff.assert_called_once_with(
             comm1.raw_id, comm2.raw_id,
             file_filter=None, opt_ignorews=False, context=3)
 
     def test_get_diff_runs_git_command_with_str_hashes(self):
         comm2 = self.repo[1]
 
         with mock.patch.object(self.repo, '_remote', return_value=mock.Mock()) as remote_mock:
             remote_mock.diff = mock.MagicMock(side_effect=callable_get_diff)
             self.repo.get_diff(self.repo.EMPTY_COMMIT, comm2)
 
         remote_mock.diff.assert_called_once_with(
             self.repo.EMPTY_COMMIT.raw_id, comm2.raw_id,
             file_filter=None, opt_ignorews=False, context=3)
 
     def test_get_diff_runs_git_command_with_path_if_its_given(self):
         comm1 = self.repo[0]
         comm2 = self.repo[1]
 
         with mock.patch.object(self.repo, '_remote', return_value=mock.Mock()) as remote_mock:
             remote_mock.diff = mock.MagicMock(side_effect=callable_get_diff)
             self.repo.get_diff(comm1, comm2, 'foo')
 
         remote_mock.diff.assert_called_once_with(
             self.repo._lookup_commit(0), comm2.raw_id,
             file_filter='foo', opt_ignorews=False, context=3)
 
 
 @pytest.mark.usefixtures("vcs_repository_support")
 class TestGitRegression(BackendTestMixin):
 
     @classmethod
     def _get_commits(cls):
         return [
             {
                 'message': 'Initial',
                 'author': 'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode(b'bot/__init__.py', content=b'base'),
                     FileNode(b'bot/templates/404.html', content=b'base'),
                     FileNode(b'bot/templates/500.html', content=b'base'),
                 ],
             },
             {
                 'message': 'Second',
                 'author': 'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'added': [
                     FileNode(b'bot/build/migrations/1.py', content=b'foo2'),
                     FileNode(b'bot/build/migrations/2.py', content=b'foo2'),
                     FileNode(b'bot/build/static/templates/f.html', content=b'foo2'),
                     FileNode(b'bot/build/static/templates/f1.html', content=b'foo2'),
                     FileNode(b'bot/build/templates/err.html', content=b'foo2'),
                     FileNode(b'bot/build/templates/err2.html', content=b'foo2'),
                 ],
             },
         ]
 
     @pytest.mark.parametrize("path, expected_paths", [
         ('bot', [
             'bot/build',
             'bot/templates',
             'bot/__init__.py']),
         ('bot/build', [
             'bot/build/migrations',
             'bot/build/static',
             'bot/build/templates']),
         ('bot/build/static', [
             'bot/build/static/templates']),
         ('bot/build/static/templates', [
             'bot/build/static/templates/f.html',
             'bot/build/static/templates/f1.html']),
         ('bot/build/templates', [
             'bot/build/templates/err.html',
             'bot/build/templates/err2.html']),
         ('bot/templates/', [
             'bot/templates/404.html',
             'bot/templates/500.html']),
     ])
     def test_similar_paths(self, path, expected_paths):
         commit = self.repo.get_commit()
         paths = [n.path for n in commit.get_nodes(path)]
         assert paths == expected_paths
 
 
 class TestDiscoverGitVersion(object):
 
     def test_returns_git_version(self, baseapp):
         version = discover_git_version()
         assert version
 
     def test_returns_empty_string_without_vcsserver(self):
         mock_connection = mock.Mock()
         mock_connection.discover_git_version = mock.Mock(
             side_effect=Exception)
         with mock.patch('rhodecode.lib.vcs.connection.Git', mock_connection):
             version = discover_git_version()
         assert version == ''
 
 
 class TestGetSubmoduleUrl(object):
     def test_submodules_file_found(self):
         commit = GitCommit(repository=mock.Mock(), raw_id='abcdef12', idx=1)
         node = mock.Mock()
 
         with mock.patch.object(
                 commit, 'get_node', return_value=node) as get_node_mock:
             node.str_content = (
                 '[submodule "subrepo1"]\n'
                 '\tpath = subrepo1\n'
                 '\turl = https://code.rhodecode.com/dulwich\n'
             )
             result = commit._get_submodule_url('subrepo1')
         get_node_mock.assert_called_once_with('.gitmodules')
         assert result == 'https://code.rhodecode.com/dulwich'
 
     def test_complex_submodule_path(self):
         commit = GitCommit(repository=mock.Mock(), raw_id='abcdef12', idx=1)
         node = mock.Mock()
 
         with mock.patch.object(
                 commit, 'get_node', return_value=node) as get_node_mock:
             node.str_content = (
                 '[submodule "complex/subrepo/path"]\n'
                 '\tpath = complex/subrepo/path\n'
                 '\turl = https://code.rhodecode.com/dulwich\n'
             )
             result = commit._get_submodule_url('complex/subrepo/path')
         get_node_mock.assert_called_once_with('.gitmodules')
         assert result == 'https://code.rhodecode.com/dulwich'
 
     def test_submodules_file_not_found(self):
         commit = GitCommit(repository=mock.Mock(), raw_id='abcdef12', idx=1)
         with mock.patch.object(
                 commit, 'get_node', side_effect=NodeDoesNotExistError):
             result = commit._get_submodule_url('complex/subrepo/path')
         assert result is None
 
     def test_path_not_found(self):
         commit = GitCommit(repository=mock.Mock(), raw_id='abcdef12', idx=1)
         node = mock.Mock()
 
         with mock.patch.object(
                 commit, 'get_node', return_value=node) as get_node_mock:
             node.str_content = (
                 '[submodule "subrepo1"]\n'
                 '\tpath = subrepo1\n'
                 '\turl = https://code.rhodecode.com/dulwich\n'
             )
             result = commit._get_submodule_url('subrepo2')
         get_node_mock.assert_called_once_with('.gitmodules')
         assert result is None
 
     def test_returns_cached_values(self):
         commit = GitCommit(repository=mock.Mock(), raw_id='abcdef12', idx=1)
         node = mock.Mock()
 
         with mock.patch.object(
                 commit, 'get_node', return_value=node) as get_node_mock:
             node.str_content = (
                 '[submodule "subrepo1"]\n'
                 '\tpath = subrepo1\n'
                 '\turl = https://code.rhodecode.com/dulwich\n'
             )
             for _ in range(3):
                 commit._get_submodule_url('subrepo1')
         get_node_mock.assert_called_once_with('.gitmodules')
 
     def test_get_node_returns_a_link(self):
         repository = mock.Mock()
         repository.alias = 'git'
         commit = GitCommit(repository=repository, raw_id='abcdef12', idx=1)
         submodule_url = 'https://code.rhodecode.com/dulwich'
         get_id_patch = mock.patch.object(
             commit, '_get_tree_id_for_path', return_value=(1, 'link'))
         get_submodule_patch = mock.patch.object(
             commit, '_get_submodule_url', return_value=submodule_url)
 
         with get_id_patch, get_submodule_patch as submodule_mock:
             node = commit.get_node('/abcde')
 
         submodule_mock.assert_called_once_with('/abcde')
         assert type(node) == SubModuleNode
         assert node.url == submodule_url
 
     def test_get_nodes_returns_links(self):
         repository = mock.MagicMock()
         repository.alias = 'git'
         repository._remote.tree_items.return_value = [
             ('subrepo', 'stat', 1, 'link')
         ]
         commit = GitCommit(repository=repository, raw_id='abcdef12', idx=1)
         submodule_url = 'https://code.rhodecode.com/dulwich'
         get_id_patch = mock.patch.object(
             commit, '_get_tree_id_for_path', return_value=(1, 'tree'))
         get_submodule_patch = mock.patch.object(
             commit, '_get_submodule_url', return_value=submodule_url)
 
         with get_id_patch, get_submodule_patch as submodule_mock:
             nodes = commit.get_nodes('/abcde')
 
         submodule_mock.assert_called_once_with('/abcde/subrepo')
         assert len(nodes) == 1
         assert type(nodes[0]) == SubModuleNode
         assert nodes[0].url == submodule_url

rhodecode/tests/vcs/test_svn.py

@@ -1,195 +1,195 b''
 
 # Copyright (C) 2010-2023 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 mock
 import pytest
 
 from rhodecode.lib.str_utils import safe_bytes
 from rhodecode.tests import SVN_REPO, TEST_DIR, TESTS_TMP_PATH
 from rhodecode.lib.vcs.backends.svn.repository import SubversionRepository
 from rhodecode.lib.vcs.conf import settings
 from rhodecode.lib.vcs.exceptions import VCSError
 
 
 pytestmark = [
     pytest.mark.backends("svn"),
     pytest.mark.usefixtures("baseapp"),
 ]
 
 
 @pytest.fixture()
 def repo(baseapp):
     repo = SubversionRepository(os.path.join(TESTS_TMP_PATH, SVN_REPO))
     return repo
 
 
 @pytest.fixture()
 def head(repo):
     return repo.get_commit()
 
 
 def test_init_fails_if_path_does_not_exist():
     path = os.path.join(TEST_DIR, 'i-do-not-exist')
     with pytest.raises(VCSError):
         SubversionRepository(path)
 
 
 def test_init_fails_if_path_is_not_a_valid_repository(tmpdir):
     path = str(tmpdir.mkdir('unicode ä'))
     with pytest.raises(VCSError):
         SubversionRepository(path)
 
 
 def test_repo_clone(vcsbackend, reposerver):
     source = vcsbackend.create_repo(number_of_commits=3)
     reposerver.serve(source)
     repo = SubversionRepository(
         vcsbackend.new_repo_path(),
         create=True,
         src_url=reposerver.url)
 
     assert source.commit_ids == repo.commit_ids
     assert source[0].message == repo[0].message
 
 
 def test_latest_commit(head):
     assert head.raw_id == '393'
 
 
 def test_commit_description(head):
     assert head.message == """Added a symlink"""
 
 
 def test_commit_author(head):
     assert head.author == 'marcin'
 
 
 @pytest.mark.parametrize("filename, content, mime_type", [
     (b'test.txt', b'Text content\n', None),
     (b'test.bin', b'\0 binary \0', 'application/octet-stream'),
 ], ids=['text', 'binary'])
 def test_sets_mime_type_correctly(vcsbackend, filename, content, mime_type):
     repo = vcsbackend.create_repo()
     vcsbackend.ensure_file(filename, content)
     file_properties = repo._remote.node_properties(filename, 1)
     assert file_properties.get('svn:mime-type') == mime_type
 
 
 def test_slice_access(repo):
     page_size = 5
     page = 0
     start = page * page_size
     end = start + page_size - 1
 
     commits = list(repo[start:end])
     assert [commit.raw_id for commit in commits] == ['1', '2', '3', '4']
 
 
 def test_walk_changelog_page(repo):
     page_size = 5
     page = 0
     start = page * page_size
     end = start + page_size - 1
 
     commits = list(repo[start:end])
     changelog = [
         'r%s, %s, %s' % (c.raw_id, c.author, c.message) for c in commits]
 
     expexted_messages = [
         'r1, marcin, initial import',
         'r2, marcin, hg ignore',
         'r3, marcin, Pip standards refactor',
         'r4, marcin, Base repository few new functions added']
     assert changelog == expexted_messages
 
 
 def test_read_full_file_tree(head):
     for topnode, dirs, files in head.walk():
         for f in files:
             len(f.content)
 
 
 def test_topnode_files_attribute(head):
     topnode = head.get_node('')
     topnode.files
 
 
 
 
 @pytest.mark.parametrize("filename, content, branch, mime_type", [
     ('branches/plain/test.txt', b'Text content\n', 'plain', None),
     ('branches/uniçö∂e/test.bin', b'\0 binary \0', 'uniçö∂e', 'application/octet-stream'),
 ], ids=['text', 'binary'])
 def test_unicode_refs(vcsbackend, filename, content, branch, mime_type):
     filename = safe_bytes(filename)
     repo = vcsbackend.create_repo()
     vcsbackend.ensure_file(filename, content)
     with mock.patch(("rhodecode.lib.vcs.backends.svn.repository"
                     ".SubversionRepository._patterns_from_section"),
                     return_value=['branches/*']):
-        assert u'branches/{0}'.format(branch) in repo.branches
+        assert f'branches/{branch}' in repo.branches
 
 
 def test_compatible_version(monkeypatch, vcsbackend):
     monkeypatch.setattr(settings, 'SVN_COMPATIBLE_VERSION', 'pre-1.8-compatible')
     path = vcsbackend.new_repo_path()
     SubversionRepository(path, create=True)
-    with open('{}/db/format'.format(path)) as f:
+    with open(f'{path}/db/format') as f:
         first_line = f.readline().strip()
         assert first_line == '4'
 
 
 def test_invalid_compatible_version(monkeypatch, vcsbackend):
     monkeypatch.setattr(settings, 'SVN_COMPATIBLE_VERSION', 'i-am-an-invalid-setting')
     path = vcsbackend.new_repo_path()
     with pytest.raises(Exception):
         SubversionRepository(path, create=True)
 
 
 class TestSVNCommit(object):
 
     @pytest.fixture(autouse=True)
     def prepare(self, repo):
         self.repo = repo
 
     def test_file_history_from_commits(self):
         node = self.repo[10].get_node('setup.py')
         commit_ids = [commit.raw_id for commit in node.history]
         assert ['8'] == commit_ids
 
         node = self.repo[20].get_node('setup.py')
         node_ids = [commit.raw_id for commit in node.history]
         assert ['18',
                 '8'] == node_ids
 
         # special case we check history from commit that has this particular
         # file changed this means we check if it's included as well
         node = self.repo.get_commit('18').get_node('setup.py')
         node_ids = [commit.raw_id for commit in node.history]
         assert ['18',
                 '8'] == node_ids
 
     def test_repo_files_content_type(self):
         test_commit = self.repo.get_commit(commit_idx=100)
         for node in test_commit.get_node('/'):
             if node.is_file():
                 assert type(node.content) == bytes
                 assert type(node.str_content) == str