tests: removed use of unicode() calls

rhodecode/tests/vcs/conftest.py

@@ -1,256 +1,256 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 time
 import shutil
 import datetime
 
 import pytest
 
 from rhodecode.lib.vcs.backends import get_backend
 from rhodecode.lib.vcs.backends.base import Config
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.tests import get_new_dir
 from rhodecode.tests.utils import check_skip_backends, check_xfail_backends
 
 
 @pytest.fixture()
 def vcs_repository_support(
         request, backend_alias, baseapp, _vcs_repo_container):
     """
     Provide a test repository for the test run.
 
     Depending on the value of `recreate_repo_per_test` a new repo for each
     test will be created.
 
     The parameter `--backends` can be used to limit this fixture to specific
     backend implementations.
     """
     cls = request.cls
 
     check_skip_backends(request.node, backend_alias)
     check_xfail_backends(request.node, backend_alias)
 
     if _should_create_repo_per_test(cls):
         _vcs_repo_container = _create_vcs_repo_container(request)
 
     repo = _vcs_repo_container.get_repo(cls, backend_alias=backend_alias)
 
     # TODO: johbo: Supporting old test class api, think about removing this
     cls.repo = repo
     cls.repo_path = repo.path
     cls.default_branch = repo.DEFAULT_BRANCH_NAME
     cls.Backend = cls.backend_class = repo.__class__
     cls.imc = repo.in_memory_commit
 
     return backend_alias, repo
 
 
 @pytest.fixture(scope='class')
 def _vcs_repo_container(request):
     """
     Internal fixture intended to help support class based scoping on demand.
     """
     return _create_vcs_repo_container(request)
 
 
 def _create_vcs_repo_container(request):
     repo_container = VcsRepoContainer()
     if not request.config.getoption('--keep-tmp-path'):
         request.addfinalizer(repo_container.cleanup)
     return repo_container
 
 
 class VcsRepoContainer(object):
 
     def __init__(self):
         self._cleanup_paths = []
         self._repos = {}
 
     def get_repo(self, test_class, backend_alias):
         if backend_alias not in self._repos:
             repo = _create_empty_repository(test_class, backend_alias)
 
             self._cleanup_paths.append(repo.path)
             self._repos[backend_alias] = repo
         return self._repos[backend_alias]
 
     def cleanup(self):
         for repo_path in reversed(self._cleanup_paths):
             shutil.rmtree(repo_path)
 
 
 def _should_create_repo_per_test(cls):
     return getattr(cls, 'recreate_repo_per_test', False)
 
 
 def _create_empty_repository(cls, backend_alias=None):
     Backend = get_backend(backend_alias or cls.backend_alias)
     repo_path = get_new_dir(str(time.time()))
     repo = Backend(repo_path, create=True)
     if hasattr(cls, '_get_commits'):
         commits = cls._get_commits()
         cls.tip = _add_commits_to_repo(repo, commits)
 
     return repo
 
 
 @pytest.fixture()
 def config():
     """
     Instance of a repository config.
 
     The instance contains only one value:
 
     - Section: "section-a"
     - Key:     "a-1"
     - Value:   "value-a-1"
 
     The intended usage is for cases where a config instance is needed but no
     specific content is required.
     """
     config = Config()
     config.set('section-a', 'a-1', 'value-a-1')
     return config
 
 
 def _add_commits_to_repo(repo, commits):
     imc = repo.in_memory_commit
     tip = None
 
     for commit in commits:
         for node in commit.get('added', []):
             imc.add(FileNode(node.path, content=node.content))
         for node in commit.get('changed', []):
             imc.change(FileNode(node.path, content=node.content))
         for node in commit.get('removed', []):
             imc.remove(FileNode(node.path))
 
         tip = imc.commit(
-            message=unicode(commit['message']),
-            author=unicode(commit['author']),
+            message=str(commit['message']),
+            author=str(commit['author']),
             date=commit['date'],
             branch=commit.get('branch'))
     return tip
 
 
 @pytest.fixture()
 def vcs_repo(request, backend_alias):
     Backend = get_backend(backend_alias)
     repo_path = get_new_dir(str(time.time()))
     repo = Backend(repo_path, create=True)
 
     @request.addfinalizer
     def cleanup():
         shutil.rmtree(repo_path)
 
     return repo
 
 
 @pytest.fixture()
 def generate_repo_with_commits(vcs_repo):
     """
     Creates a fabric to generate N comits with some file nodes on a randomly
     generated repository
     """
 
     def commit_generator(num):
         start_date = datetime.datetime(2010, 1, 1, 20)
         for x in range(num):
             yield {
                 'message': 'Commit %d' % x,
                 'author': 'Joe Doe <[e-mail]>',
                 'date': start_date + datetime.timedelta(hours=12 * x),
                 'added': [
                     FileNode('file_%d.txt' % x, content='Foobar %d' % x),
                 ],
                 'modified': [
                     FileNode('file_%d.txt' % x,
                              content='Foobar %d modified' % (x-1)),
                 ]
             }
 
     def commit_maker(num=5):
         _add_commits_to_repo(vcs_repo, commit_generator(num))
         return vcs_repo
 
     return commit_maker
 
 
 @pytest.fixture()
 def hg_repo(request, vcs_repo):
     repo = vcs_repo
 
     commits = repo._get_commits()
     _add_commits_to_repo(repo, commits)
 
     return repo
 
 
 @pytest.fixture()
 def hg_commit(hg_repo):
     return hg_repo.get_commit()
 
 
 class BackendTestMixin(object):
     """
     This is a backend independent test case class which should be created
     with ``type`` method.
 
     It is required to set following attributes at subclass:
 
     - ``backend_alias``: alias of used backend (see ``vcs.BACKENDS``)
     - ``repo_path``: path to the repository which would be created for set of
       tests
     - ``recreate_repo_per_test``: If set to ``False``, repo would NOT be
       created
       before every single test. Defaults to ``True``.
     """
     recreate_repo_per_test = True
 
     @classmethod
     def _get_commits(cls):
         commits = [
             {
                 'message': u'Initial commit',
                 'author': u'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('foobar', content='Foobar'),
                     FileNode('foobar2', content='Foobar II'),
                     FileNode('foo/bar/baz', content='baz here!'),
                 ],
             },
             {
                 'message': u'Changes...',
                 'author': u'Jane Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 21),
                 'added': [
                     FileNode('some/new.txt', content='news...'),
                 ],
                 'changed': [
                     FileNode('foobar', 'Foobar I'),
                 ],
                 'removed': [],
             },
         ]
         return commits
 
 

rhodecode/tests/vcs/test_repository.py

@@ -1,552 +1,552 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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
 from urllib.error import URLError
 
 import mock
 import pytest
 
 from rhodecode.lib.vcs import backends
 from rhodecode.lib.vcs.backends.base import (
     Config, BaseInMemoryCommit, Reference, MergeResponse, MergeFailureReason)
 from rhodecode.lib.vcs.exceptions import VCSError, RepositoryError
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.tests.vcs.conftest import BackendTestMixin
 from rhodecode.tests import repo_id_generator
 
 
 @pytest.mark.usefixtures("vcs_repository_support")
 class TestRepositoryBase(BackendTestMixin):
     recreate_repo_per_test = False
 
     def test_init_accepts_unicode_path(self, tmpdir):
-        path = unicode(tmpdir.join(u'unicode ä'))
+        path = str(tmpdir.join(u'unicode ä'))
         self.Backend(path, create=True)
 
     def test_init_accepts_str_path(self, tmpdir):
         path = str(tmpdir.join('str ä'))
         self.Backend(path, create=True)
 
     def test_init_fails_if_path_does_not_exist(self, tmpdir):
-        path = unicode(tmpdir.join('i-do-not-exist'))
+        path = str(tmpdir.join('i-do-not-exist'))
         with pytest.raises(VCSError):
             self.Backend(path)
 
     def test_init_fails_if_path_is_not_a_valid_repository(self, tmpdir):
-        path = unicode(tmpdir.mkdir(u'unicode ä'))
+        path = str(tmpdir.mkdir(u'unicode ä'))
         with pytest.raises(VCSError):
             self.Backend(path)
 
     def test_has_commits_attribute(self):
         self.repo.commit_ids
 
     def test_name(self):
         assert self.repo.name.startswith('vcs-test')
 
     @pytest.mark.backends("hg", "git")
     def test_has_default_branch_name(self):
         assert self.repo.DEFAULT_BRANCH_NAME is not None
 
     @pytest.mark.backends("svn")
     def test_has_no_default_branch_name(self):
         assert self.repo.DEFAULT_BRANCH_NAME is None
 
     def test_has_empty_commit(self):
         assert self.repo.EMPTY_COMMIT_ID is not None
         assert self.repo.EMPTY_COMMIT is not None
 
     def test_empty_changeset_is_deprecated(self):
         def get_empty_changeset(repo):
             return repo.EMPTY_CHANGESET
         pytest.deprecated_call(get_empty_changeset, self.repo)
 
     def test_bookmarks(self):
         assert len(self.repo.bookmarks) == 0
 
     # TODO: Cover two cases: Local repo path, remote URL
     def test_check_url(self):
         config = Config()
         assert self.Backend.check_url(self.repo.path, config)
 
     def test_check_url_invalid(self):
         config = Config()
         with pytest.raises(URLError):
             self.Backend.check_url(self.repo.path + "invalid", config)
 
     def test_get_contact(self):
         assert self.repo.contact
 
     def test_get_description(self):
         assert self.repo.description
 
     def test_get_hook_location(self):
         assert len(self.repo.get_hook_location()) != 0
 
     def test_last_change(self, local_dt_to_utc):
         assert self.repo.last_change >= local_dt_to_utc(
             datetime.datetime(2010, 1, 1, 21, 0))
 
     def test_last_change_in_empty_repository(self, vcsbackend, local_dt_to_utc):
         delta = datetime.timedelta(seconds=1)
 
         start = local_dt_to_utc(datetime.datetime.now())
         empty_repo = vcsbackend.create_repo()
         now = local_dt_to_utc(datetime.datetime.now())
         assert empty_repo.last_change >= start - delta
         assert empty_repo.last_change <= now + delta
 
     def test_repo_equality(self):
         assert self.repo == self.repo
 
     def test_repo_equality_broken_object(self):
         import copy
         _repo = copy.copy(self.repo)
         delattr(_repo, 'path')
         assert self.repo != _repo
 
     def test_repo_equality_other_object(self):
         class dummy(object):
             path = self.repo.path
         assert self.repo != dummy()
 
     def test_get_commit_is_implemented(self):
         self.repo.get_commit()
 
     def test_get_commits_is_implemented(self):
         commit_iter = iter(self.repo.get_commits())
         commit = next(commit_iter)
         assert commit.idx == 0
 
     def test_supports_iteration(self):
         repo_iter = iter(self.repo)
         commit = next(repo_iter)
         assert commit.idx == 0
 
     def test_in_memory_commit(self):
         imc = self.repo.in_memory_commit
         assert isinstance(imc, BaseInMemoryCommit)
 
     @pytest.mark.backends("hg")
     def test__get_url_unicode(self):
         url = u'/home/repos/malmö'
         assert self.repo._get_url(url)
 
 
 @pytest.mark.usefixtures("vcs_repository_support")
 class TestDeprecatedRepositoryAPI(BackendTestMixin):
     recreate_repo_per_test = False
 
     def test_revisions_is_deprecated(self):
         def get_revisions(repo):
             return repo.revisions
         pytest.deprecated_call(get_revisions, self.repo)
 
     def test_get_changeset_is_deprecated(self):
         pytest.deprecated_call(self.repo.get_changeset)
 
     def test_get_changesets_is_deprecated(self):
         pytest.deprecated_call(self.repo.get_changesets)
 
     def test_in_memory_changeset_is_deprecated(self):
         def get_imc(repo):
             return repo.in_memory_changeset
         pytest.deprecated_call(get_imc, self.repo)
 
 
 # TODO: these tests are incomplete, must check the resulting compare result for
 # correcteness
 class TestRepositoryCompare:
 
     @pytest.mark.parametrize('merge', [True, False])
     def test_compare_commits_of_same_repository(self, vcsbackend, merge):
         target_repo = vcsbackend.create_repo(number_of_commits=5)
         target_repo.compare(
             target_repo[1].raw_id, target_repo[3].raw_id, target_repo,
             merge=merge)
 
     @pytest.mark.xfail_backends('svn')
     @pytest.mark.parametrize('merge', [True, False])
     def test_compare_cloned_repositories(self, vcsbackend, merge):
         target_repo = vcsbackend.create_repo(number_of_commits=5)
         source_repo = vcsbackend.clone_repo(target_repo)
         assert target_repo != source_repo
 
         vcsbackend.add_file(source_repo, 'newfile', 'somecontent')
         source_commit = source_repo.get_commit()
 
         target_repo.compare(
             target_repo[1].raw_id, source_repo[3].raw_id, source_repo,
             merge=merge)
 
     @pytest.mark.xfail_backends('svn')
     @pytest.mark.parametrize('merge', [True, False])
     def test_compare_unrelated_repositories(self, vcsbackend, merge):
         orig = vcsbackend.create_repo(number_of_commits=5)
         unrelated = vcsbackend.create_repo(number_of_commits=5)
         assert orig != unrelated
 
         orig.compare(
             orig[1].raw_id, unrelated[3].raw_id, unrelated, merge=merge)
 
 
 class TestRepositoryGetCommonAncestor:
 
     def test_get_common_ancestor_from_same_repo_existing(self, vcsbackend):
         target_repo = vcsbackend.create_repo(number_of_commits=5)
 
         expected_ancestor = target_repo[2].raw_id
 
         assert target_repo.get_common_ancestor(
             commit_id1=target_repo[2].raw_id,
             commit_id2=target_repo[4].raw_id,
             repo2=target_repo
         ) == expected_ancestor
 
         assert target_repo.get_common_ancestor(
             commit_id1=target_repo[4].raw_id,
             commit_id2=target_repo[2].raw_id,
             repo2=target_repo
         ) == expected_ancestor
 
     @pytest.mark.xfail_backends("svn")
     def test_get_common_ancestor_from_cloned_repo_existing(self, vcsbackend):
         target_repo = vcsbackend.create_repo(number_of_commits=5)
         source_repo = vcsbackend.clone_repo(target_repo)
         assert target_repo != source_repo
 
         vcsbackend.add_file(source_repo, 'newfile', 'somecontent')
         source_commit = source_repo.get_commit()
 
         expected_ancestor = target_repo[4].raw_id
 
         assert target_repo.get_common_ancestor(
             commit_id1=target_repo[4].raw_id,
             commit_id2=source_commit.raw_id,
             repo2=source_repo
         ) == expected_ancestor
 
         assert target_repo.get_common_ancestor(
             commit_id1=source_commit.raw_id,
             commit_id2=target_repo[4].raw_id,
             repo2=target_repo
         ) == expected_ancestor
 
     @pytest.mark.xfail_backends("svn")
     def test_get_common_ancestor_from_unrelated_repo_missing(self, vcsbackend):
         original = vcsbackend.create_repo(number_of_commits=5)
         unrelated = vcsbackend.create_repo(number_of_commits=5)
         assert original != unrelated
 
         assert original.get_common_ancestor(
             commit_id1=original[0].raw_id,
             commit_id2=unrelated[0].raw_id,
             repo2=unrelated
         ) is None
 
         assert original.get_common_ancestor(
             commit_id1=original[-1].raw_id,
             commit_id2=unrelated[-1].raw_id,
             repo2=unrelated
         ) is None
 
 
 @pytest.mark.backends("git", "hg")
 class TestRepositoryMerge(object):
     def prepare_for_success(self, vcsbackend):
         self.target_repo = vcsbackend.create_repo(number_of_commits=1)
         self.source_repo = vcsbackend.clone_repo(self.target_repo)
         vcsbackend.add_file(self.target_repo, 'README_MERGE1', 'Version 1')
         vcsbackend.add_file(self.source_repo, 'README_MERGE2', 'Version 2')
         imc = self.source_repo.in_memory_commit
         imc.add(FileNode('file_x', content=self.source_repo.name))
         imc.commit(
             message=u'Automatic commit from repo merge test',
             author=u'Automatic <[e-mail]>')
         self.target_commit = self.target_repo.get_commit()
         self.source_commit = self.source_repo.get_commit()
         # This only works for Git and Mercurial
         default_branch = self.target_repo.DEFAULT_BRANCH_NAME
         self.target_ref = Reference('branch', default_branch, self.target_commit.raw_id)
         self.source_ref = Reference('branch', default_branch, self.source_commit.raw_id)
         self.workspace_id = 'test-merge-{}'.format(vcsbackend.alias)
         self.repo_id = repo_id_generator(self.target_repo.path)
 
     def prepare_for_conflict(self, vcsbackend):
         self.target_repo = vcsbackend.create_repo(number_of_commits=1)
         self.source_repo = vcsbackend.clone_repo(self.target_repo)
         vcsbackend.add_file(self.target_repo, 'README_MERGE', 'Version 1')
         vcsbackend.add_file(self.source_repo, 'README_MERGE', 'Version 2')
         self.target_commit = self.target_repo.get_commit()
         self.source_commit = self.source_repo.get_commit()
         # This only works for Git and Mercurial
         default_branch = self.target_repo.DEFAULT_BRANCH_NAME
         self.target_ref = Reference('branch', default_branch, self.target_commit.raw_id)
         self.source_ref = Reference('branch', default_branch, self.source_commit.raw_id)
         self.workspace_id = 'test-merge-{}'.format(vcsbackend.alias)
         self.repo_id = repo_id_generator(self.target_repo.path)
 
     def test_merge_success(self, vcsbackend):
         self.prepare_for_success(vcsbackend)
 
         merge_response = self.target_repo.merge(
             self.repo_id, self.workspace_id, self.target_ref, self.source_repo,
             self.source_ref,
             'test user', '[e-mail]', 'merge message 1',
             dry_run=False)
         expected_merge_response = MergeResponse(
             True, True, merge_response.merge_ref,
             MergeFailureReason.NONE)
         assert merge_response == expected_merge_response
 
         target_repo = backends.get_backend(vcsbackend.alias)(
             self.target_repo.path)
         target_commits = list(target_repo.get_commits())
         commit_ids = [c.raw_id for c in target_commits[:-1]]
         assert self.source_ref.commit_id in commit_ids
         assert self.target_ref.commit_id in commit_ids
 
         merge_commit = target_commits[-1]
         assert merge_commit.raw_id == merge_response.merge_ref.commit_id
         assert merge_commit.message.strip() == 'merge message 1'
         assert merge_commit.author == 'test user <[e-mail]>'
 
         # We call it twice so to make sure we can handle updates
         target_ref = Reference(
             self.target_ref.type, self.target_ref.name,
             merge_response.merge_ref.commit_id)
 
         merge_response = target_repo.merge(
             self.repo_id, self.workspace_id, target_ref, self.source_repo, self.source_ref,
             'test user', '[e-mail]', 'merge message 2',
             dry_run=False)
         expected_merge_response = MergeResponse(
             True, True, merge_response.merge_ref,
             MergeFailureReason.NONE)
         assert merge_response == expected_merge_response
 
         target_repo = backends.get_backend(
             vcsbackend.alias)(self.target_repo.path)
         merge_commit = target_repo.get_commit(
             merge_response.merge_ref.commit_id)
         assert merge_commit.message.strip() == 'merge message 1'
         assert merge_commit.author == 'test user <[e-mail]>'
 
     def test_merge_success_dry_run(self, vcsbackend):
         self.prepare_for_success(vcsbackend)
 
         merge_response = self.target_repo.merge(
             self.repo_id, self.workspace_id, self.target_ref, self.source_repo,
             self.source_ref, dry_run=True)
 
         # We call it twice so to make sure we can handle updates
         merge_response_update = self.target_repo.merge(
             self.repo_id, self.workspace_id, self.target_ref, self.source_repo,
             self.source_ref, dry_run=True)
 
         # Multiple merges may differ in their commit id. Therefore we set the
         # commit id to `None` before comparing the merge responses.
         new_merge_ref = merge_response.merge_ref._replace(commit_id=None)
         merge_response.merge_ref = new_merge_ref
 
         new_update_merge_ref = merge_response_update.merge_ref._replace(commit_id=None)
         merge_response_update.merge_ref = new_update_merge_ref
 
         assert merge_response == merge_response_update
         assert merge_response.possible is True
         assert merge_response.executed is False
         assert merge_response.merge_ref
         assert merge_response.failure_reason is MergeFailureReason.NONE
 
     @pytest.mark.parametrize('dry_run', [True, False])
     def test_merge_conflict(self, vcsbackend, dry_run):
         self.prepare_for_conflict(vcsbackend)
 
         expected_merge_response = MergeResponse(
             False, False, None, MergeFailureReason.MERGE_FAILED)
 
         merge_response = self.target_repo.merge(
             self.repo_id,  self.workspace_id, self.target_ref,
             self.source_repo, self.source_ref,
             'test_user', '[e-mail]', 'test message', dry_run=dry_run)
         assert merge_response == expected_merge_response
 
         # We call it twice so to make sure we can handle updates
         merge_response = self.target_repo.merge(
             self.repo_id, self.workspace_id, self.target_ref, self.source_repo,
             self.source_ref,
             'test_user', '[e-mail]', 'test message', dry_run=dry_run)
         assert merge_response == expected_merge_response
 
     def test_merge_target_is_not_head(self, vcsbackend):
         self.prepare_for_success(vcsbackend)
         target_ref = Reference(
             self.target_ref.type, self.target_ref.name, '0' * 40)
         expected_merge_response = MergeResponse(
             False, False, None, MergeFailureReason.TARGET_IS_NOT_HEAD,
             metadata={'target_ref': target_ref})
         merge_response = self.target_repo.merge(
             self.repo_id, self.workspace_id, target_ref, self.source_repo,
             self.source_ref, dry_run=True)
 
         assert merge_response == expected_merge_response
 
     def test_merge_missing_source_reference(self, vcsbackend):
         self.prepare_for_success(vcsbackend)
 
         source_ref = Reference(
             self.source_ref.type, 'not_existing', self.source_ref.commit_id)
         expected_merge_response = MergeResponse(
             False, False, None, MergeFailureReason.MISSING_SOURCE_REF,
             metadata={'source_ref': source_ref})
 
         merge_response = self.target_repo.merge(
             self.repo_id, self.workspace_id, self.target_ref,
             self.source_repo, source_ref,
             dry_run=True)
 
         assert merge_response == expected_merge_response
 
     def test_merge_raises_exception(self, vcsbackend):
         self.prepare_for_success(vcsbackend)
         expected_merge_response = MergeResponse(
             False, False, None, MergeFailureReason.UNKNOWN,
             metadata={'exception': 'ErrorForTest'})
 
         with mock.patch.object(self.target_repo, '_merge_repo',
                                side_effect=RepositoryError()):
             merge_response = self.target_repo.merge(
                 self.repo_id, self.workspace_id, self.target_ref,
                 self.source_repo, self.source_ref,
                 dry_run=True)
 
         assert merge_response == expected_merge_response
 
     def test_merge_invalid_user_name(self, vcsbackend):
         repo = vcsbackend.create_repo(number_of_commits=1)
         ref = Reference('branch', 'master', 'not_used')
         workspace_id = 'test-errors-in-merge'
         repo_id = repo_id_generator(workspace_id)
         with pytest.raises(ValueError):
             repo.merge(repo_id,  workspace_id, ref, self, ref)
 
     def test_merge_invalid_user_email(self, vcsbackend):
         repo = vcsbackend.create_repo(number_of_commits=1)
         ref = Reference('branch', 'master', 'not_used')
         workspace_id = 'test-errors-in-merge'
         repo_id = repo_id_generator(workspace_id)
         with pytest.raises(ValueError):
             repo.merge(
                 repo_id,  workspace_id, ref, self, ref, 'user name')
 
     def test_merge_invalid_message(self, vcsbackend):
         repo = vcsbackend.create_repo(number_of_commits=1)
         ref = Reference('branch', 'master', 'not_used')
         workspace_id = 'test-errors-in-merge'
         repo_id = repo_id_generator(workspace_id)
         with pytest.raises(ValueError):
             repo.merge(
                 repo_id, workspace_id, ref, self, ref,
                 'user name', '[e-mail]')
 
 
 @pytest.mark.usefixtures("vcs_repository_support")
 class TestRepositoryStrip(BackendTestMixin):
     recreate_repo_per_test = True
 
     @classmethod
     def _get_commits(cls):
         commits = [
             {
                 'message': 'Initial commit',
                 'author': 'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'branch': 'master',
                 'added': [
                     FileNode('foobar', content='foobar'),
                     FileNode('foobar2', content='foobar2'),
                 ],
             },
         ]
         for x in range(10):
             commit_data = {
                 'message': 'Changed foobar - commit%s' % x,
                 'author': 'Jane Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 21, x),
                 'branch': 'master',
                 'changed': [
                     FileNode('foobar', 'FOOBAR - %s' % x),
                 ],
             }
             commits.append(commit_data)
         return commits
 
     @pytest.mark.backends("git", "hg")
     def test_strip_commit(self):
         tip = self.repo.get_commit()
         assert tip.idx == 10
         self.repo.strip(tip.raw_id, self.repo.DEFAULT_BRANCH_NAME)
 
         tip = self.repo.get_commit()
         assert tip.idx == 9
 
     @pytest.mark.backends("git", "hg")
     def test_strip_multiple_commits(self):
         tip = self.repo.get_commit()
         assert tip.idx == 10
 
         old = self.repo.get_commit(commit_idx=5)
         self.repo.strip(old.raw_id, self.repo.DEFAULT_BRANCH_NAME)
 
         tip = self.repo.get_commit()
         assert tip.idx == 4
 
 
 @pytest.mark.backends('hg', 'git')
 class TestRepositoryPull(object):
 
     def test_pull(self, vcsbackend):
         source_repo = vcsbackend.repo
         target_repo = vcsbackend.create_repo()
         assert len(source_repo.commit_ids) > len(target_repo.commit_ids)
 
         target_repo.pull(source_repo.path)
         # Note: Get a fresh instance, avoids caching trouble
         target_repo = vcsbackend.backend(target_repo.path)
         assert len(source_repo.commit_ids) == len(target_repo.commit_ids)
 
     def test_pull_wrong_path(self, vcsbackend):
         target_repo = vcsbackend.create_repo()
         with pytest.raises(RepositoryError):
             target_repo.pull(target_repo.path + "wrong")
 
     def test_pull_specific_commits(self, vcsbackend):
         source_repo = vcsbackend.repo
         target_repo = vcsbackend.create_repo()
 
         second_commit = source_repo[1].raw_id
         if vcsbackend.alias == 'git':
             second_commit_ref = 'refs/test-refs/a'
             source_repo.set_refs(second_commit_ref, second_commit)
 
         target_repo.pull(source_repo.path, commit_ids=[second_commit])
         target_repo = vcsbackend.backend(target_repo.path)
         assert 2 == len(target_repo.commit_ids)
         assert second_commit == target_repo.get_commit().raw_id

rhodecode/tests/vcs/test_svn.py

@@ -1,186 +1,186 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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.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 = unicode(tmpdir.mkdir(u'unicode ä'))
+    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", [
     ('test.txt', 'Text content\n', None),
     ('test.bin', '\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", [
     (u'branches/plain/test.txt', 'Text content\n', 'plain', None),
     (u'branches/uniçö∂e/test.bin', '\0 binary \0', u'uniçö∂e',
         'application/octet-stream'),
 ], ids=['text', 'binary'])
 def test_unicode_refs(vcsbackend, filename, content, branch, mime_type):
     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
 
 
 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:
         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