diff --git a/rhodecode/tests/vcs/test_repository.py b/rhodecode/tests/vcs/test_repository.py --- a/rhodecode/tests/vcs/test_repository.py +++ b/rhodecode/tests/vcs/test_repository.py @@ -1,4 +1,3 @@ - # Copyright (C) 2010-2023 RhodeCode GmbH # # This program is free software: you can redistribute it and/or modify @@ -25,7 +24,12 @@ import pytest from rhodecode.lib.vcs import backends from rhodecode.lib.vcs.backends.base import ( - Config, BaseInMemoryCommit, Reference, MergeResponse, MergeFailureReason) + 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 @@ -37,20 +41,20 @@ class TestRepositoryBase(BackendTestMixi recreate_repo_per_test = False def test_init_accepts_unicode_path(self, tmpdir): - path = str(tmpdir.join(u'unicode ä')) + path = str(tmpdir.join("unicode ä")) self.Backend(path, create=True) def test_init_accepts_str_path(self, tmpdir): - path = str(tmpdir.join('str ä')) + path = str(tmpdir.join("str ä")) self.Backend(path, create=True) def test_init_fails_if_path_does_not_exist(self, tmpdir): - path = str(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 = str(tmpdir.mkdir(u'unicode ä')) + path = str(tmpdir.mkdir("unicode ä")) with pytest.raises(VCSError): self.Backend(path) @@ -58,7 +62,7 @@ class TestRepositoryBase(BackendTestMixi assert self.repo.commit_ids def test_name(self): - assert self.repo.name.startswith('vcs-test') + assert self.repo.name.startswith("vcs-test") @pytest.mark.backends("hg", "git") def test_has_default_branch_name(self): @@ -75,6 +79,7 @@ class TestRepositoryBase(BackendTestMixi 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): @@ -87,9 +92,9 @@ class TestRepositoryBase(BackendTestMixi def test_check_url_on_remote_url(self): config = Config() url = { - 'hg': 'https://code.rhodecode.com/rhodecode-vcsserver', - 'svn': 'https://code.rhodecode.com/svn-doc', - 'git': 'https://code.rhodecode.com/appenlight', + "hg": "https://code.rhodecode.com/rhodecode-vcsserver", + "svn": "https://code.rhodecode.com/svn-doc", + "git": "https://code.rhodecode.com/appenlight", }[self.repo.alias] assert self.Backend.check_url(url, config) @@ -110,7 +115,8 @@ class TestRepositoryBase(BackendTestMixi 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)) + 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) @@ -126,13 +132,15 @@ class TestRepositoryBase(BackendTestMixi def test_repo_equality_broken_object(self): import copy + _repo = copy.copy(self.repo) - delattr(_repo, 'path') + 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): @@ -154,7 +162,7 @@ class TestRepositoryBase(BackendTestMixi @pytest.mark.backends("hg") def test__get_url_unicode(self): - url = u'/home/repos/malmö' + url = "/home/repos/malmö" assert self.repo._get_url(url) @@ -165,6 +173,7 @@ class TestDeprecatedRepositoryAPI(Backen 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): @@ -176,63 +185,67 @@ class TestDeprecatedRepositoryAPI(Backen 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]) + @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) + 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]) + @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, b'newfile', b'somecontent') + vcsbackend.add_file(source_repo, b"newfile", b"somecontent") source_commit = source_repo.get_commit() target_repo.compare( - target_repo[1].raw_id, source_repo[3].raw_id, source_repo, - merge=merge) + 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]) + @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) + 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[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 + 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): @@ -240,22 +253,28 @@ class TestRepositoryGetCommonAncestor: source_repo = vcsbackend.clone_repo(target_repo) assert target_repo != source_repo - vcsbackend.add_file(source_repo, b'newfile', b'somecontent') + vcsbackend.add_file(source_repo, b"newfile", b"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=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 + 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): @@ -263,17 +282,23 @@ class TestRepositoryGetCommonAncestor: 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[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 + 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") @@ -281,51 +306,57 @@ 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, b'README_MERGE1', b'Version 1') - vcsbackend.add_file(self.source_repo, b'README_MERGE2', b'Version 2') + vcsbackend.add_file(self.target_repo, b"README_MERGE1", b"Version 1") + vcsbackend.add_file(self.source_repo, b"README_MERGE2", b"Version 2") imc = self.source_repo.in_memory_commit - imc.add(FileNode(b'file_x', content=self.source_repo.name)) + imc.add(FileNode(b"file_x", content=self.source_repo.name)) imc.commit( - message=u'Automatic commit from repo merge test', - author=u'Automatic ') + message="Automatic commit from repo merge test", + author="Automatic ", + ) 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.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, b'README_MERGE', b'Version 1') - vcsbackend.add_file(self.source_repo, b'README_MERGE', b'Version 2') + vcsbackend.add_file(self.target_repo, b"README_MERGE", b"Version 1") + vcsbackend.add_file(self.source_repo, b"README_MERGE", b"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.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.repo_id, + self.workspace_id, + self.target_ref, + self.source_repo, self.source_ref, - 'test user', 'test@rhodecode.com', 'merge message 1', - dry_run=False) + "test user", + "test@rhodecode.com", + "merge message 1", + dry_run=False, + ) expected_merge_response = MergeResponse( - True, True, merge_response.merge_ref, - MergeFailureReason.NONE) + 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_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 @@ -333,49 +364,67 @@ class TestRepositoryMerge(object): 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 ' + assert merge_commit.message.strip() == "merge message 1" + assert merge_commit.author == "test user " # 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) + 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', 'test@rhodecode.com', 'merge message 2', - dry_run=False) + self.repo_id, + self.workspace_id, + target_ref, + self.source_repo, + self.source_ref, + "test user", + "test@rhodecode.com", + "merge message 2", + dry_run=False, + ) expected_merge_response = MergeResponse( - True, True, merge_response.merge_ref, - MergeFailureReason.NONE) + 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 ' + 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 " 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) + 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) + self.repo_id, + self.workspace_id, + self.target_ref, + self.source_repo, + self.source_ref, + dry_run=True, + ) + + assert merge_response.merge_ref + assert merge_response_update.merge_ref # 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.commit_id = None - merge_response.merge_ref = new_merge_ref + merge_response.merge_ref.commit_id = 'abcdeabcde' - new_update_merge_ref = merge_response_update.merge_ref.commit_id = None - merge_response_update.merge_ref = new_update_merge_ref + merge_response_update.merge_ref.commit_id = 'abcdeabcde' assert merge_response == merge_response_update assert merge_response.possible is True @@ -383,36 +432,59 @@ class TestRepositoryMerge(object): assert merge_response.merge_ref assert merge_response.failure_reason is MergeFailureReason.NONE - @pytest.mark.parametrize('dry_run', [True, False]) + @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) + 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', 'test@rhodecode.com', 'test message', dry_run=dry_run) + self.repo_id, + self.workspace_id, + self.target_ref, + self.source_repo, + self.source_ref, + "test_user", + "test@rhodecode.com", + "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.repo_id, + self.workspace_id, + self.target_ref, + self.source_repo, self.source_ref, - 'test_user', 'test@rhodecode.com', 'test message', dry_run=dry_run) + "test_user", + "test@rhodecode.com", + "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) + 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}) + 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) + self.repo_id, + self.workspace_id, + target_ref, + self.source_repo, + self.source_ref, + dry_run=True, + ) assert merge_response == expected_merge_response @@ -420,59 +492,76 @@ class TestRepositoryMerge(object): self.prepare_for_success(vcsbackend) source_ref = Reference( - self.source_ref.type, 'not_existing', self.source_ref.commit_id) + 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}) + 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) + 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'}) + False, + False, + None, + MergeFailureReason.UNKNOWN, + metadata={"exception": "ErrorForTest"}, + ) - with mock.patch.object(self.target_repo, '_merge_repo', - side_effect=RepositoryError()): + 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) + 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' + 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) + 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' + 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') - - 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', 'user@email.com') + repo_id, workspace_id, ref, self, ref, "user name", "user@email.com" + ) @pytest.mark.usefixtures("vcs_repository_support") @@ -483,24 +572,24 @@ class TestRepositoryStrip(BackendTestMix def _get_commits(cls): commits = [ { - 'message': 'Initial commit', - 'author': 'Joe Doe ', - 'date': datetime.datetime(2010, 1, 1, 20), - 'branch': 'master', - 'added': [ - FileNode(b'foobar', content='foobar'), - FileNode(b'foobar2', content='foobar2'), + "message": "Initial commit", + "author": "Joe Doe ", + "date": datetime.datetime(2010, 1, 1, 20), + "branch": "master", + "added": [ + FileNode(b"foobar", content="foobar"), + FileNode(b"foobar2", content="foobar2"), ], }, ] for x in range(10): commit_data = { - 'message': 'Changed foobar - commit%s' % x, - 'author': 'Jane Doe ', - 'date': datetime.datetime(2010, 1, 1, 21, x), - 'branch': 'master', - 'changed': [ - FileNode(b'foobar', 'FOOBAR - %s' % x), + "message": "Changed foobar - commit%s" % x, + "author": "Jane Doe ", + "date": datetime.datetime(2010, 1, 1, 21, x), + "branch": "master", + "changed": [ + FileNode(b"foobar", "FOOBAR - %s" % x), ], } commits.append(commit_data) @@ -527,9 +616,8 @@ class TestRepositoryStrip(BackendTestMix assert tip.idx == 4 -@pytest.mark.backends('hg', 'git') +@pytest.mark.backends("hg", "git") class TestRepositoryPull(object): - def test_pull(self, vcsbackend): source_repo = vcsbackend.repo target_repo = vcsbackend.create_repo() @@ -550,8 +638,8 @@ class TestRepositoryPull(object): target_repo = vcsbackend.create_repo() second_commit = source_repo[1].raw_id - if vcsbackend.alias == 'git': - second_commit_ref = 'refs/test-refs/a' + 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])