# Copyright (C) 2010-2024 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 . # # 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.lib.utils import make_db_config from rhodecode.lib.vcs import backends from rhodecode.lib.vcs.backends.base import Reference, MergeResponse, MergeFailureReason from rhodecode.lib.vcs.backends.hg import MercurialRepository, MercurialCommit from rhodecode.lib.vcs.exceptions import RepositoryError, VCSError, NodeDoesNotExistError, CommitDoesNotExistError from rhodecode.lib.vcs.nodes import FileNode, NodeKind, NodeState from rhodecode.tests import TEST_HG_REPO, TEST_HG_REPO_CLONE, repo_id_generator pytestmark = pytest.mark.backends("hg") def repo_path_generator(): """ Return a different path to be used for cloning repos. """ i = 0 while True: i += 1 yield "%s-%d" % (TEST_HG_REPO_CLONE, i) REPO_PATH_GENERATOR = repo_path_generator() @pytest.fixture(scope="class", autouse=True) def repo(request, baseapp): repo = MercurialRepository(TEST_HG_REPO) if request.cls: request.cls.repo = repo return repo class TestMercurialRepository(object): # pylint: disable=protected-access def get_clone_repo(self): """ Return a clone of the base repo. """ clone_path = next(REPO_PATH_GENERATOR) repo_clone = MercurialRepository(clone_path, create=True, src_url=self.repo.path) return repo_clone def get_empty_repo(self): """ Return an empty repo. """ return MercurialRepository(next(REPO_PATH_GENERATOR), create=True) def test_wrong_repo_path(self): wrong_repo_path = "/tmp/errorrepo_hg" with pytest.raises(RepositoryError): MercurialRepository(wrong_repo_path) def test_unicode_path_repo(self): with pytest.raises(VCSError): MercurialRepository("iShouldFail") def test_unicode_commit_id(self): with pytest.raises(CommitDoesNotExistError): self.repo.get_commit("unicode-commit-id") with pytest.raises(CommitDoesNotExistError): self.repo.get_commit("unícøde-spéçial-chäråcter-commit-id") def test_unicode_bookmark(self): self.repo.bookmark("unicode-bookmark") self.repo.bookmark("unícøde-spéçial-chäråcter-bookmark") def test_unicode_branch(self): with pytest.raises(KeyError): assert self.repo.branches["unicode-branch"] with pytest.raises(KeyError): assert self.repo.branches["unícøde-spéçial-chäråcter-branch"] def test_repo_clone(self): if os.path.exists(TEST_HG_REPO_CLONE): self.fail( "Cannot test mercurial clone repo as location %s already " "exists. You should manually remove it first." % TEST_HG_REPO_CLONE ) repo = MercurialRepository(TEST_HG_REPO) repo_clone = MercurialRepository(TEST_HG_REPO_CLONE, create=True, src_url=TEST_HG_REPO) 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_with_update(self): repo = MercurialRepository(TEST_HG_REPO) repo_clone = MercurialRepository( TEST_HG_REPO_CLONE + "_w_update", create=True, src_url=TEST_HG_REPO, do_workspace_checkout=True ) assert len(repo.commit_ids) == len(repo_clone.commit_ids) # check if current workdir was updated assert os.path.isfile(os.path.join(TEST_HG_REPO_CLONE + "_w_update", "MANIFEST.in")) def test_repo_clone_without_update(self): repo = MercurialRepository(TEST_HG_REPO) repo_clone = MercurialRepository( TEST_HG_REPO_CLONE + "_wo_update", create=True, src_url=TEST_HG_REPO, do_workspace_checkout=False ) assert len(repo.commit_ids) == len(repo_clone.commit_ids) assert not os.path.isfile(os.path.join(TEST_HG_REPO_CLONE + "_wo_update", "MANIFEST.in")) def test_commit_ids(self): # there are 21 commits at bitbucket now # so we can assume they would be available from now on subset = { "b986218ba1c9b0d6a259fac9b050b1724ed8e545", "3d8f361e72ab303da48d799ff1ac40d5ac37c67e", "6cba7170863a2411822803fa77a0a264f1310b35", "56349e29c2af3ac913b28bde9a2c6154436e615b", "2dda4e345facb0ccff1a191052dd1606dba6781d", "6fff84722075f1607a30f436523403845f84cd9e", "7d4bc8ec6be56c0f10425afb40b6fc315a4c25e7", "3803844fdbd3b711175fc3da9bdacfcd6d29a6fb", "dc5d2c0661b61928834a785d3e64a3f80d3aad9c", "be90031137367893f1c406e0a8683010fd115b79", "db8e58be770518cbb2b1cdfa69146e47cd481481", "84478366594b424af694a6c784cb991a16b87c21", "17f8e105dddb9f339600389c6dc7175d395a535c", "20a662e756499bde3095ffc9bc0643d1def2d0eb", "2e319b85e70a707bba0beff866d9f9de032aa4f9", "786facd2c61deb9cf91e9534735124fb8fc11842", "94593d2128d38210a2fcd1aabff6dda0d6d9edf8", "aa6a0de05b7612707db567078e130a6cd114a9a7", "eada5a770da98ab0dd7325e29d00e0714f228d09", } assert subset.issubset(set(self.repo.commit_ids)) # check if we have the proper order of commits org = [ "b986218ba1c9b0d6a259fac9b050b1724ed8e545", "3d8f361e72ab303da48d799ff1ac40d5ac37c67e", "6cba7170863a2411822803fa77a0a264f1310b35", "56349e29c2af3ac913b28bde9a2c6154436e615b", "2dda4e345facb0ccff1a191052dd1606dba6781d", "6fff84722075f1607a30f436523403845f84cd9e", "7d4bc8ec6be56c0f10425afb40b6fc315a4c25e7", "3803844fdbd3b711175fc3da9bdacfcd6d29a6fb", "dc5d2c0661b61928834a785d3e64a3f80d3aad9c", "be90031137367893f1c406e0a8683010fd115b79", "db8e58be770518cbb2b1cdfa69146e47cd481481", "84478366594b424af694a6c784cb991a16b87c21", "17f8e105dddb9f339600389c6dc7175d395a535c", "20a662e756499bde3095ffc9bc0643d1def2d0eb", "2e319b85e70a707bba0beff866d9f9de032aa4f9", "786facd2c61deb9cf91e9534735124fb8fc11842", "94593d2128d38210a2fcd1aabff6dda0d6d9edf8", "aa6a0de05b7612707db567078e130a6cd114a9a7", "eada5a770da98ab0dd7325e29d00e0714f228d09", "2c1885c735575ca478bf9e17b0029dca68824458", "d9bcd465040bf869799b09ad732c04e0eea99fe9", "469e9c847fe1f6f7a697b8b25b4bc5b48780c1a7", "4fb8326d78e5120da2c7468dcf7098997be385da", "62b4a097164940bd66030c4db51687f3ec035eed", "536c1a19428381cfea92ac44985304f6a8049569", "965e8ab3c44b070cdaa5bf727ddef0ada980ecc4", "9bb326a04ae5d98d437dece54be04f830cf1edd9", "f8940bcb890a98c4702319fbe36db75ea309b475", "ff5ab059786ebc7411e559a2cc309dfae3625a3b", "6b6ad5f82ad5bb6190037671bd254bd4e1f4bf08", "ee87846a61c12153b51543bf860e1026c6d3dcba", ] assert org == self.repo.commit_ids[:31] def test_iter_slice(self): sliced = list(self.repo[:10]) itered = list(self.repo)[:10] assert sliced == itered 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)]: indexes = list(self.repo[sfrom:sto]) assert len(indexes) == size assert indexes[0] == self.repo.get_commit(commit_idx=sfrom) assert indexes[-1] == self.repo.get_commit(commit_idx=sto - 1) def test_branches(self): # TODO: Need more tests here # active branches assert "default" in self.repo.branches assert "stable" in self.repo.branches # closed assert "git" in self.repo._get_branches(closed=True) assert "web" in self.repo._get_branches(closed=True) for name, id in self.repo.branches.items(): assert isinstance(self.repo.get_commit(id), MercurialCommit) def test_tip_in_tags(self): # tip is always a tag assert "tip" in self.repo.tags def test_tip_commit_in_tags(self): tip = self.repo.get_commit() assert self.repo.tags["tip"] == tip.raw_id def test_initial_commit(self): init_commit = self.repo.get_commit(commit_idx=0) init_author = init_commit.author assert init_commit.message == "initial import" assert init_author == "Marcin Kuzminski " assert init_author == init_commit.committer assert sorted(init_commit._file_paths) == sorted( [ "vcs/__init__.py", "vcs/backends/BaseRepository.py", "vcs/backends/__init__.py", ] ) assert sorted(init_commit._dir_paths) == sorted(["", "vcs", "vcs/backends"]) assert init_commit._dir_paths + init_commit._file_paths == init_commit._paths 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): # rawid with pytest.raises(RepositoryError): self.repo.get_commit("abcd" * 10) # shortid with pytest.raises(RepositoryError): self.repo.get_commit("erro" * 4) # numeric with pytest.raises(RepositoryError): self.repo.get_commit(commit_idx=self.repo.count() + 1) # Small chance we ever get to this one idx = pow(2, 30) with pytest.raises(RepositoryError): self.repo.get_commit(commit_idx=idx) def test_commit10(self): commit10 = self.repo.get_commit(commit_idx=10) 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_local_clone(self): clone_path = next(REPO_PATH_GENERATOR) self.repo._local_clone(clone_path) repo_clone = MercurialRepository(clone_path) assert self.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) def test_update(self): repo_clone = self.get_clone_repo() branches = repo_clone.branches repo_clone._update("default") assert branches["default"] == repo_clone._identify() repo_clone._update("stable") assert branches["stable"] == repo_clone._identify() def test_local_pull_branch(self): target_repo = self.get_empty_repo() source_repo = self.get_clone_repo() default = Reference("branch", "default", source_repo.branches["default"]) target_repo._local_pull(source_repo.path, default) target_repo = MercurialRepository(target_repo.path) assert target_repo.branches["default"] == source_repo.branches["default"] stable = Reference("branch", "stable", source_repo.branches["stable"]) target_repo._local_pull(source_repo.path, stable) target_repo = MercurialRepository(target_repo.path) assert target_repo.branches["stable"] == source_repo.branches["stable"] def test_local_pull_bookmark(self): target_repo = self.get_empty_repo() source_repo = self.get_clone_repo() commits = list(source_repo.get_commits(branch_name="default")) foo1_id = commits[-5].raw_id foo1 = Reference("book", "foo1", foo1_id) source_repo._update(foo1_id) source_repo.bookmark("foo1") foo2_id = commits[-3].raw_id foo2 = Reference("book", "foo2", foo2_id) source_repo._update(foo2_id) source_repo.bookmark("foo2") target_repo._local_pull(source_repo.path, foo1) target_repo = MercurialRepository(target_repo.path) assert target_repo.branches["default"] == commits[-5].raw_id target_repo._local_pull(source_repo.path, foo2) target_repo = MercurialRepository(target_repo.path) assert target_repo.branches["default"] == commits[-3].raw_id def test_local_pull_commit(self): target_repo = self.get_empty_repo() source_repo = self.get_clone_repo() commits = list(source_repo.get_commits(branch_name="default")) commit_id = commits[-5].raw_id commit = Reference("rev", commit_id, commit_id) target_repo._local_pull(source_repo.path, commit) target_repo = MercurialRepository(target_repo.path) assert target_repo.branches["default"] == commit_id commit_id = commits[-3].raw_id commit = Reference("rev", commit_id, commit_id) target_repo._local_pull(source_repo.path, commit) target_repo = MercurialRepository(target_repo.path) assert target_repo.branches["default"] == commit_id def test_local_pull_from_same_repo(self): reference = Reference("branch", "default", None) with pytest.raises(ValueError): self.repo._local_pull(self.repo.path, reference) def test_validate_pull_reference_raises_on_missing_reference(self, vcsbackend_hg): target_repo = vcsbackend_hg.create_repo(number_of_commits=1) reference = Reference("book", "invalid_reference", "a" * 40) with pytest.raises(CommitDoesNotExistError): target_repo._validate_pull_reference(reference) def test_heads(self): assert set(self.repo._heads()) == set(self.repo.branches.values()) def test_ancestor(self): commits = [c.raw_id for c in self.repo.get_commits(branch_name="default")] assert self.repo._ancestor(commits[-3], commits[-5]) == commits[-5] assert self.repo._ancestor(commits[-5], commits[-3]) == commits[-5] def test_local_push(self): target_repo = self.get_empty_repo() revisions = list(self.repo.get_commits(branch_name="default")) revision = revisions[-5].raw_id self.repo._local_push(revision, target_repo.path) target_repo = MercurialRepository(target_repo.path) assert target_repo.branches["default"] == revision def test_hooks_can_be_enabled_for_local_push(self): revision = "deadbeef" repo_path = "test_group/test_repo" with mock.patch.object(self.repo, "_remote") as remote_mock: self.repo._local_push(revision, repo_path, enable_hooks=True) remote_mock.push.assert_called_once_with([revision], repo_path, hooks=True, push_branches=False) def test_local_merge(self, vcsbackend_hg): target_repo = vcsbackend_hg.create_repo(number_of_commits=1) source_repo = vcsbackend_hg.clone_repo(target_repo) vcsbackend_hg.add_file(target_repo, b"README_MERGE1", b"Version 1") target_repo = MercurialRepository(target_repo.path) target_rev = target_repo.branches["default"] target_ref = Reference(type="branch", name="default", commit_id=target_rev) vcsbackend_hg.add_file(source_repo, b"README_MERGE2", b"Version 2") source_repo = MercurialRepository(source_repo.path) source_rev = source_repo.branches["default"] source_ref = Reference(type="branch", name="default", commit_id=source_rev) target_repo._local_pull(source_repo.path, source_ref) merge_message = "Merge message\n\nDescription:..." user_name = "Albert Einstein" user_email = "albert@einstein.com" merge_commit_id, needs_push = target_repo._local_merge( target_ref, merge_message, user_name, user_email, source_ref ) assert needs_push target_repo = MercurialRepository(target_repo.path) assert target_repo.commit_ids[-3] == target_rev assert target_repo.commit_ids[-2] == source_rev last_commit = target_repo.get_commit(merge_commit_id) 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, ".hg", "merge", "state")) def test_local_merge_source_is_fast_forward(self, vcsbackend_hg): target_repo = vcsbackend_hg.create_repo(number_of_commits=1) source_repo = vcsbackend_hg.clone_repo(target_repo) target_rev = target_repo.branches["default"] target_ref = Reference(type="branch", name="default", commit_id=target_rev) vcsbackend_hg.add_file(source_repo, "README_MERGE2", "Version 2") source_repo = MercurialRepository(source_repo.path) source_rev = source_repo.branches["default"] source_ref = Reference(type="branch", name="default", commit_id=source_rev) target_repo._local_pull(source_repo.path, source_ref) merge_message = "Merge message\n\nDescription:..." user_name = "Albert Einstein" user_email = "albert@einstein.com" merge_commit_id, needs_push = target_repo._local_merge( target_ref, merge_message, user_name, user_email, source_ref ) assert merge_commit_id == source_rev assert needs_push target_repo = MercurialRepository(target_repo.path) assert target_repo.commit_ids[-2] == target_rev assert target_repo.commit_ids[-1] == source_rev assert not os.path.exists(os.path.join(target_repo.path, ".hg", "merge", "state")) def test_local_merge_source_is_integrated(self, vcsbackend_hg): target_repo = vcsbackend_hg.create_repo(number_of_commits=1) target_rev = target_repo.branches["default"] target_ref = Reference(type="branch", name="default", commit_id=target_rev) merge_message = "Merge message\n\nDescription:..." user_name = "Albert Einstein" user_email = "albert@einstein.com" merge_commit_id, needs_push = target_repo._local_merge( target_ref, merge_message, user_name, user_email, target_ref ) assert merge_commit_id == target_rev assert not needs_push target_repo = MercurialRepository(target_repo.path) assert target_repo.commit_ids[-1] == target_rev assert not os.path.exists(os.path.join(target_repo.path, ".hg", "merge", "state")) def test_local_merge_raises_exception_on_conflict(self, vcsbackend_hg): target_repo = vcsbackend_hg.create_repo(number_of_commits=1) source_repo = vcsbackend_hg.clone_repo(target_repo) vcsbackend_hg.add_file(target_repo, "README_MERGE", "Version 1") target_repo = MercurialRepository(target_repo.path) target_rev = target_repo.branches["default"] target_ref = Reference(type="branch", name="default", commit_id=target_rev) vcsbackend_hg.add_file(source_repo, "README_MERGE", "Version 2") source_repo = MercurialRepository(source_repo.path) source_rev = source_repo.branches["default"] source_ref = Reference(type="branch", name="default", commit_id=source_rev) target_repo._local_pull(source_repo.path, source_ref) with pytest.raises(RepositoryError): target_repo._local_merge(target_ref, "merge_message", "user name", "user@name.com", source_ref) # Check we are not left in an intermediate merge state assert not os.path.exists(os.path.join(target_repo.path, ".hg", "merge", "state")) def test_local_merge_of_two_branches_of_the_same_repo(self, backend_hg): commits = [ {"message": "a"}, {"message": "b", "branch": "b"}, ] repo = backend_hg.create_repo(commits) commit_ids = backend_hg.commit_ids target_ref = Reference(type="branch", name="default", commit_id=commit_ids["a"]) source_ref = Reference(type="branch", name="b", commit_id=commit_ids["b"]) merge_message = "Merge message\n\nDescription:..." user_name = "Albert Einstein" user_email = "albert@einstein.com" vcs_repo = repo.scm_instance() merge_commit_id, needs_push = vcs_repo._local_merge( target_ref, merge_message, user_name, user_email, source_ref ) assert merge_commit_id != source_ref.commit_id assert needs_push is True commit = vcs_repo.get_commit(merge_commit_id) assert commit.merge is True assert commit.message == merge_message def test_maybe_prepare_merge_workspace(self): workspace = self.repo._maybe_prepare_merge_workspace(1, "pr2", "unused", "unused2") assert os.path.isdir(workspace) workspace_repo = MercurialRepository(workspace) assert workspace_repo.branches == self.repo.branches # Calling it a second time should also succeed workspace = self.repo._maybe_prepare_merge_workspace(1, "pr2", "unused", "unused2") assert os.path.isdir(workspace) def test_cleanup_merge_workspace(self): workspace = self.repo._maybe_prepare_merge_workspace(1, "pr3", "unused", "unused2") assert os.path.isdir(workspace) self.repo.cleanup_merge_workspace(1, "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_merge_target_is_bookmark(self, vcsbackend_hg): target_repo = vcsbackend_hg.create_repo(number_of_commits=1) source_repo = vcsbackend_hg.clone_repo(target_repo) vcsbackend_hg.add_file(target_repo, "README_MERGE1", "Version 1") vcsbackend_hg.add_file(source_repo, "README_MERGE2", "Version 2") imc = source_repo.in_memory_commit imc.add(FileNode(b"file_x", content=source_repo.name)) imc.commit(message="Automatic commit from repo merge test", author="Automatic ") target_commit = target_repo.get_commit() source_commit = source_repo.get_commit() default_branch = target_repo.DEFAULT_BRANCH_NAME bookmark_name = "bookmark" target_repo._update(default_branch) target_repo.bookmark(bookmark_name) target_ref = Reference("book", bookmark_name, target_commit.raw_id) source_ref = Reference("branch", default_branch, source_commit.raw_id) workspace_id = "test-merge" repo_id = repo_id_generator(target_repo.path) merge_response = target_repo.merge( repo_id, workspace_id, target_ref, source_repo, source_ref, "test user", "test@rhodecode.com", "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_hg.alias)(target_repo.path) target_commits = list(target_repo.get_commits()) commit_ids = [c.raw_id for c in target_commits[:-1]] assert source_ref.commit_id in commit_ids assert 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 " # Check the bookmark was updated in the target repo assert target_repo.bookmarks[bookmark_name] == merge_response.merge_ref.commit_id def test_merge_source_is_bookmark(self, vcsbackend_hg): target_repo = vcsbackend_hg.create_repo(number_of_commits=1) source_repo = vcsbackend_hg.clone_repo(target_repo) imc = source_repo.in_memory_commit imc.add(FileNode(b"file_x", content=source_repo.name)) imc.commit(message="Automatic commit from repo merge test", author="Automatic ") target_commit = target_repo.get_commit() source_commit = source_repo.get_commit() default_branch = target_repo.DEFAULT_BRANCH_NAME bookmark_name = "bookmark" target_ref = Reference("branch", default_branch, target_commit.raw_id) source_repo._update(default_branch) source_repo.bookmark(bookmark_name) source_ref = Reference("book", bookmark_name, source_commit.raw_id) workspace_id = "test-merge" repo_id = repo_id_generator(target_repo.path) merge_response = target_repo.merge( repo_id, workspace_id, target_ref, source_repo, source_ref, "test user", "test@rhodecode.com", "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_hg.alias)(target_repo.path) target_commits = list(target_repo.get_commits()) commit_ids = [c.raw_id for c in target_commits] assert source_ref.commit_id == commit_ids[-1] assert target_ref.commit_id == commit_ids[-2] def test_merge_target_has_multiple_heads(self, vcsbackend_hg): target_repo = vcsbackend_hg.create_repo(number_of_commits=2) source_repo = vcsbackend_hg.clone_repo(target_repo) vcsbackend_hg.add_file(target_repo, "README_MERGE1", "Version 1") vcsbackend_hg.add_file(source_repo, "README_MERGE2", "Version 2") # add an extra head to the target repo imc = target_repo.in_memory_commit imc.add(FileNode(b"file_x", content="foo")) commits = list(target_repo.get_commits()) imc.commit( message="Automatic commit from repo merge test", author="Automatic ", parents=commits[0:1], ) target_commit = target_repo.get_commit() source_commit = source_repo.get_commit() default_branch = target_repo.DEFAULT_BRANCH_NAME target_repo._update(default_branch) target_ref = Reference("branch", default_branch, target_commit.raw_id) source_ref = Reference("branch", default_branch, source_commit.raw_id) workspace_id = "test-merge" assert len(target_repo._heads(branch="default")) == 2 heads = target_repo._heads(branch="default") expected_merge_response = MergeResponse( False, False, None, MergeFailureReason.HG_TARGET_HAS_MULTIPLE_HEADS, metadata={"heads": heads} ) repo_id = repo_id_generator(target_repo.path) merge_response = target_repo.merge( repo_id, workspace_id, target_ref, source_repo, source_ref, "test user", "test@rhodecode.com", "merge message 1", dry_run=False, ) assert merge_response == expected_merge_response def test_merge_rebase_source_is_updated_bookmark(self, vcsbackend_hg): target_repo = vcsbackend_hg.create_repo(number_of_commits=1) source_repo = vcsbackend_hg.clone_repo(target_repo) vcsbackend_hg.add_file(target_repo, b"README_MERGE1", b"Version 1") vcsbackend_hg.add_file(source_repo, b"README_MERGE2", b"Version 2") imc = source_repo.in_memory_commit imc.add(FileNode(b"file_x", content=safe_bytes(source_repo.name))) imc.commit(message="Automatic commit from repo merge test", author="Automatic ") target_commit = target_repo.get_commit() source_commit = source_repo.get_commit() vcsbackend_hg.add_file(source_repo, b"LICENSE", b"LICENSE Info") default_branch = target_repo.DEFAULT_BRANCH_NAME bookmark_name = "bookmark" source_repo._update(default_branch) source_repo.bookmark(bookmark_name) target_ref = Reference("branch", default_branch, target_commit.raw_id) source_ref = Reference("book", bookmark_name, source_commit.raw_id) repo_id = repo_id_generator(target_repo.path) workspace_id = "test-merge" merge_response = target_repo.merge( repo_id, workspace_id, target_ref, source_repo, source_ref, "test user", "test@rhodecode.com", "merge message 1", dry_run=False, use_rebase=True, ) expected_merge_response = MergeResponse(True, True, merge_response.merge_ref, MergeFailureReason.NONE) assert merge_response == expected_merge_response target_repo = backends.get_backend(vcsbackend_hg.alias)(target_repo.path) last_commit = target_repo.get_commit() assert last_commit.message == source_commit.message assert last_commit.author == source_commit.author # This checks that we effectively did a rebase assert last_commit.raw_id != source_commit.raw_id # Check the target has only 4 commits: 2 were already in target and # only two should have been added assert len(target_repo.commit_ids) == 2 + 2 class TestGetShadowInstance(object): @pytest.fixture() def repo(self, vcsbackend_hg): _hg_repo = vcsbackend_hg.repo connection_mock = mock.Mock(unsafe=True, name="connection.Hg") mock.patch("rhodecode.lib.vcs.connection.Hg", connection_mock) return _hg_repo def test_getting_shadow_instance_copies_config(self, repo): shadow = repo.get_shadow_instance(repo.path) assert shadow.config.serialize() == repo.config.serialize() def test_disables_hooks_section(self, repo): repo.config.set('hooks', 'foo', 'val') shadow = repo.get_shadow_instance(repo.path) assert not shadow.config.items('hooks') def test_allows_to_keep_hooks(self, repo): repo.config.set('hooks', 'foo', 'val') shadow = repo.get_shadow_instance(repo.path, enable_hooks=True) assert shadow.config.items('hooks') class TestMercurialCommit(object): def _test_equality(self, commit): idx = commit.idx assert commit == self.repo.get_commit(commit_idx=idx) def test_equality(self): indexes = [0, 10, 20] commits = [self.repo.get_commit(commit_idx=idx) for idx in indexes] for commit in commits: self._test_equality(commit) def test_default_commit(self): tip = self.repo.get_commit("tip") assert tip == self.repo.get_commit() assert tip == self.repo.get_commit(commit_id=None) assert tip == self.repo.get_commit(commit_idx=None) assert tip == list(self.repo[-1:])[0] def test_root_node(self): tip = self.repo.get_commit("tip") 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 = self.repo.get_commit(commit_idx=45) 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 = self.repo.get_commit(commit_idx=45) 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): commit0 = self.repo.get_commit(commit_idx=0) assert commit0.branch == "default" assert commit0.tags == [] commit10 = self.repo.get_commit(commit_idx=10) assert commit10.branch == "default" assert commit10.tags == [] commit44 = self.repo.get_commit(commit_idx=44) assert commit44.branch == "web" tip = self.repo.get_commit("tip") assert "tip" in tip.tags def test_bookmarks(self): commit0 = self.repo.get_commit(commit_idx=0) assert commit0.bookmarks == [] def _test_file_size(self, idx, path, size): node = self.repo.get_commit(commit_idx=idx).get_node(path) assert node.is_file() assert node.size == size def test_file_size(self): to_check = ( (10, "setup.py", 1068), (20, "setup.py", 1106), (60, "setup.py", 1074), (10, "vcs/backends/base.py", 2921), (20, "vcs/backends/base.py", 3936), (60, "vcs/backends/base.py", 6189), ) for idx, path, size in to_check: self._test_file_size(idx, path, 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 ["3803844fdbd3b711175fc3da9bdacfcd6d29a6fb"] == commit_ids node = self.repo[20].get_node("setup.py") node_ids = [commit.raw_id for commit in node.history] assert ["eada5a770da98ab0dd7325e29d00e0714f228d09", "3803844fdbd3b711175fc3da9bdacfcd6d29a6fb"] == 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("eada5a770da98ab0dd7325e29d00e0714f228d09").get_node("setup.py") node_ids = [commit.raw_id for commit in node.history] assert ["eada5a770da98ab0dd7325e29d00e0714f228d09", "3803844fdbd3b711175fc3da9bdacfcd6d29a6fb"] == 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": [7, 18, 45, 46, 47, 69, 77], "vcs/nodes.py": [7, 8, 24, 26, 30, 45, 47, 49, 56, 57, 58, 59, 60, 61, 73, 76], "vcs/backends/hg.py": [ 4, 5, 6, 11, 12, 13, 14, 15, 16, 21, 22, 23, 26, 27, 28, 30, 31, 33, 35, 36, 37, 38, 39, 40, 41, 44, 45, 47, 48, 49, 53, 54, 55, 58, 60, 61, 67, 68, 69, 70, 73, 77, 78, 79, 82, ], } for path, indexes in files.items(): tip = self.repo.get_commit(commit_idx=indexes[-1]) node = tip.get_node(path) node_indexes = [commit.idx for commit in node.history] assert set(indexes).issubset(set(node_indexes)), ( "We assumed that %s is subset of commits for which file %s " "has been changed, and history of that node returned: %s" % (indexes, path, node_indexes) ) def test_file_annotate(self): files = { "vcs/backends/__init__.py": { 89: { "lines_no": 31, "commits": [ 32, 32, 61, 32, 32, 37, 32, 32, 32, 44, 37, 37, 37, 37, 45, 37, 44, 37, 37, 37, 32, 32, 32, 32, 37, 32, 37, 37, 32, 32, 32, ], }, 20: {"lines_no": 1, "commits": [4]}, 55: { "lines_no": 31, "commits": [ 32, 32, 45, 32, 32, 37, 32, 32, 32, 44, 37, 37, 37, 37, 45, 37, 44, 37, 37, 37, 32, 32, 32, 32, 37, 32, 37, 37, 32, 32, 32, ], }, }, "vcs/exceptions.py": { 89: { "lines_no": 18, "commits": [16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 17, 16, 16, 18, 18, 18], }, 20: { "lines_no": 18, "commits": [16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 17, 16, 16, 18, 18, 18], }, 55: { "lines_no": 18, "commits": [16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 16, 17, 16, 16, 18, 18, 18], }, }, "MANIFEST.in": { 89: {"lines_no": 5, "commits": [7, 7, 7, 71, 71]}, 20: {"lines_no": 3, "commits": [7, 7, 7]}, 55: {"lines_no": 3, "commits": [7, 7, 7]}, }, } for fname, commit_dict in files.items(): for idx, __ in commit_dict.items(): commit = self.repo.get_commit(commit_idx=idx) 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_2 = [x[2]().idx for x in commit.get_file_annotate(fname)] l2 = files[fname][idx]["commits"] assert l1 == l2, ( "The lists of commit for %s@commit_id%s" "from annotation list should match each other," "got \n%s \nvs \n%s " % (fname, idx, l1, l2) ) def test_commit_state(self): """ Tests which files have been added/changed/removed at particular commit """ # commit_id 46ad32a4f974: # hg st --rev 46ad32a4f974 # changed: 13 # added: 20 # removed: 1 changed = set( [ ".hgignore", "README.rst", "docs/conf.py", "docs/index.rst", "setup.py", "tests/test_hg.py", "tests/test_nodes.py", "vcs/__init__.py", "vcs/backends/__init__.py", "vcs/backends/base.py", "vcs/backends/hg.py", "vcs/nodes.py", "vcs/utils/__init__.py", ] ) added = set( [ "docs/api/backends/hg.rst", "docs/api/backends/index.rst", "docs/api/index.rst", "docs/api/nodes.rst", "docs/api/web/index.rst", "docs/api/web/simplevcs.rst", "docs/installation.rst", "docs/quickstart.rst", "setup.cfg", "vcs/utils/baseui_config.py", "vcs/utils/web.py", "vcs/web/__init__.py", "vcs/web/exceptions.py", "vcs/web/simplevcs/__init__.py", "vcs/web/simplevcs/exceptions.py", "vcs/web/simplevcs/middleware.py", "vcs/web/simplevcs/models.py", "vcs/web/simplevcs/settings.py", "vcs/web/simplevcs/utils.py", "vcs/web/simplevcs/views.py", ] ) removed = set(["docs/api.rst"]) commit64 = self.repo.get_commit("46ad32a4f974") assert set((node.path for node in commit64.added)) == added assert set((node.path for node in commit64.changed)) == changed assert set((node.path for node in commit64.removed)) == removed # commit_id b090f22d27d6: # hg st --rev b090f22d27d6 # changed: 13 # added: 20 # removed: 1 commit88 = self.repo.get_commit("b090f22d27d6") assert set((node.path for node in commit88.added)) == set() assert set((node.path for node in commit88.changed)) == set([".hgignore"]) assert set((node.path for node in commit88.removed)) == set() # # 85: # added: 2 [ # 'vcs/utils/diffs.py', 'vcs/web/simplevcs/views/diffs.py'] # changed: 4 ['vcs/web/simplevcs/models.py', ...] # removed: 1 ['vcs/utils/web.py'] commit85 = self.repo.get_commit(commit_idx=85) assert set((node.path for node in commit85.added)) == set( ["vcs/utils/diffs.py", "vcs/web/simplevcs/views/diffs.py"] ) assert set((node.path for node in commit85.changed)) == set( [ "vcs/web/simplevcs/models.py", "vcs/web/simplevcs/utils.py", "vcs/web/simplevcs/views/__init__.py", "vcs/web/simplevcs/views/repository.py", ] ) assert set((node.path for node in commit85.removed)) == set(["vcs/utils/web.py"]) def test_files_state(self): """ Tests state of FileNodes. """ commit = self.repo.get_commit(commit_idx=85) node = commit.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 commit = self.repo.get_commit(commit_idx=88) node = commit.get_node(".hgignore") assert node.state, NodeState.CHANGED assert not node.added assert node.changed assert not node.not_changed assert not node.removed commit = self.repo.get_commit(commit_idx=85) node = commit.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(commit_idx=2) 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] def test_commit_message_is_unicode(self): for cm in self.repo: assert type(cm.message) == str def test_commit_author_is_unicode(self): for cm in self.repo: assert type(cm.author) == str 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 def test_wrong_path(self): # There is 'setup.py' in the root dir but not there: path = "foo/bar/setup.py" with pytest.raises(VCSError): self.repo.get_commit().get_node(path) def test_author_email(self): assert "marcin@python-blog.com" == self.repo.get_commit("b986218ba1c9").author_email assert "lukasz.balcerzak@python-center.pl" == self.repo.get_commit("3803844fdbd3").author_email assert "" == self.repo.get_commit("84478366594b").author_email def test_author_username(self): assert "Marcin Kuzminski" == self.repo.get_commit("b986218ba1c9").author_name assert "Lukasz Balcerzak" == self.repo.get_commit("3803844fdbd3").author_name assert "marcink" == self.repo.get_commit("84478366594b").author_name class TestLargeFileRepo(object): def test_large_file(self, backend_hg): conf = make_db_config() hg_largefiles_store = conf.get("largefiles", "usercache") repo = backend_hg.create_test_repo("largefiles", conf) tip = repo.scm_instance().get_commit() node = tip.get_node(".hglf/thisfileislarge") lf_node = node.get_largefile_node() assert lf_node.is_largefile() is True assert lf_node.size == 1024000 assert lf_node.name == ".hglf/thisfileislarge" class TestGetBranchName(object): def test_returns_ref_name_when_type_is_branch(self): ref = self._create_ref("branch", "fake-name") result = self.repo._get_branch_name(ref) assert result == ref.name @pytest.mark.parametrize("type_", ("book", "tag")) def test_queries_remote_when_type_is_not_branch(self, type_): ref = self._create_ref(type_, "wrong-fake-name") with mock.patch.object(self.repo, "_remote") as remote_mock: remote_mock.ctx_branch.return_value = "fake-name" result = self.repo._get_branch_name(ref) assert result == "fake-name" remote_mock.ctx_branch.assert_called_once_with(ref.commit_id) def _create_ref(self, type_, name): ref = mock.Mock() ref.type = type_ ref.name = "wrong-fake-name" ref.commit_id = "deadbeef" return ref class TestIsTheSameBranch(object): def test_returns_true_when_branches_are_equal(self): source_ref = mock.Mock(name="source-ref") target_ref = mock.Mock(name="target-ref") branch_name_patcher = mock.patch.object(self.repo, "_get_branch_name", return_value="default") with branch_name_patcher as branch_name_mock: result = self.repo._is_the_same_branch(source_ref, target_ref) expected_calls = [mock.call(source_ref), mock.call(target_ref)] assert branch_name_mock.call_args_list == expected_calls assert result is True def test_returns_false_when_branches_are_not_equal(self): source_ref = mock.Mock(name="source-ref") source_ref.name = "source-branch" target_ref = mock.Mock(name="target-ref") source_ref.name = "target-branch" def side_effect(ref): return ref.name branch_name_patcher = mock.patch.object(self.repo, "_get_branch_name", side_effect=side_effect) with branch_name_patcher as branch_name_mock: result = self.repo._is_the_same_branch(source_ref, target_ref) expected_calls = [mock.call(source_ref), mock.call(target_ref)] assert branch_name_mock.call_args_list == expected_calls assert result is False