tests: fixed tests for new source ref checking.

rhodecode/tests/vcs/test_git.py

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

rhodecode/tests/vcs/test_hg.py

@@ -1,1180 +1,1183 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import os
 
 import mock
 import pytest
 
 from rhodecode.lib.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
 
 
 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:
 
     # 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(u'iShouldFail')
 
     def test_unicode_commit_id(self):
         with pytest.raises(CommitDoesNotExistError):
             self.repo.get_commit(u'unicode-commit-id')
         with pytest.raises(CommitDoesNotExistError):
             self.repo.get_commit(u'unícøde-spéçial-chäråcter-commit-id')
 
     def test_unicode_bookmark(self):
         self.repo.bookmark(u'unicode-bookmark')
         self.repo.bookmark(u'unícøde-spéçial-chäråcter-bookmark')
 
     def test_unicode_branch(self):
         with pytest.raises(KeyError):
             self.repo.branches[u'unicode-branch']
         with pytest.raises(KeyError):
             self.repo.branches[u'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,
                                          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',
             src_url=TEST_HG_REPO, update_after_clone=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',
             src_url=TEST_HG_REPO, update_after_clone=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 = set([
             '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 <[e-mail]>'
         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.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, 'README_MERGE1', '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_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 = '[e-mail]'
         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 = '[e-mail]'
         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 = '[e-mail]'
         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', '[e-mail]',
                 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 = '[e-mail]'
         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('pr2', 'unused')
+        workspace = self.repo._maybe_prepare_merge_workspace(
+            '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('pr2', 'unused')
+        workspace = self.repo._maybe_prepare_merge_workspace(
+            'pr2', 'unused', 'unused2')
         assert os.path.isdir(workspace)
 
     def test_cleanup_merge_workspace(self):
-        workspace = self.repo._maybe_prepare_merge_workspace('pr3', 'unused')
+        workspace = self.repo._maybe_prepare_merge_workspace(
+            'pr3', 'unused', 'unused2')
         self.repo.cleanup_merge_workspace('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('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('file_x', content=source_repo.name))
         imc.commit(
             message=u'Automatic commit from repo merge test',
             author=u'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 = 'test-merge'
 
         merge_response = target_repo.merge(
             target_ref, source_repo, source_ref, workspace,
             '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_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 <[e-mail]>'
 
         # 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('file_x', content=source_repo.name))
         imc.commit(
             message=u'Automatic commit from repo merge test',
             author=u'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 = 'test-merge'
 
         merge_response = target_repo.merge(
             target_ref, source_repo, source_ref, workspace,
             '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_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('file_x', content='foo'))
         commits = list(target_repo.get_commits())
         imc.commit(
             message=u'Automatic commit from repo merge test',
             author=u'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 = 'test-merge'
 
         assert len(target_repo._heads(branch='default')) == 2
         expected_merge_response = MergeResponse(
             False, False, None,
             MergeFailureReason.HG_TARGET_HAS_MULTIPLE_HEADS)
         merge_response = target_repo.merge(
             target_ref, source_repo, source_ref, workspace,
             'test user', '[e-mail]', '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, 'README_MERGE1', 'Version 1')
         vcsbackend_hg.add_file(source_repo, 'README_MERGE2', 'Version 2')
         imc = source_repo.in_memory_commit
         imc.add(FileNode('file_x', content=source_repo.name))
         imc.commit(
             message=u'Automatic commit from repo merge test',
             author=u'Automatic')
         target_commit = target_repo.get_commit()
         source_commit = source_repo.get_commit()
 
         vcsbackend_hg.add_file(source_repo, 'LICENSE', '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)
         workspace = 'test-merge'
 
         merge_response = target_repo.merge(
             target_ref, source_repo, source_ref, workspace,
             'test user', '[e-mail]', '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, monkeypatch):
         repo = vcsbackend_hg.repo
         monkeypatch.setattr(repo, 'config', mock.Mock())
         monkeypatch.setattr('rhodecode.lib.vcs.connection.Hg', mock.Mock())
         return repo
 
     def test_passes_config(self, repo):
         shadow = repo._get_shadow_instance(repo.path)
         assert shadow.config == repo.config.copy()
 
     def test_disables_hooks(self, repo):
         shadow = repo._get_shadow_instance(repo.path)
         shadow.config.clear_section.assert_called_once_with('hooks')
 
     def test_allows_to_keep_hooks(self, repo):
         shadow = repo._get_shadow_instance(repo.path, enable_hooks=True)
         assert not shadow.config.clear_section.called
 
 
 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) == unicode
 
     def test_commit_author_is_unicode(self):
         for cm in self.repo:
             assert type(cm.author) == unicode
 
     def test_repo_files_content_is_unicode(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) == unicode
 
     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 '[e-mail]' == \
             self.repo.get_commit('b986218ba1c9').author_email
         assert '[e-mail]' == \
             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):
         repo = backend_hg.create_test_repo('largefiles', make_db_config())
 
         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

rhodecode/tests/vcs/test_hg_vcsserver_cache_invalidation.py

@@ -1,183 +1,183 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import pytest
 
 from mock import call, patch
 
 from rhodecode.lib.vcs.backends.base import Reference
 
 
 class TestMercurialRemoteRepoInvalidation(object):
     """
     If the VCSServer is running with multiple processes or/and instances.
     Operations on repositories are potentially handled by different processes
     in a random fashion. The mercurial repository objects used in the VCSServer
     are caching the commits of the repo. Therefore we have to invalidate the
     VCSServer caching of these objects after a writing operation.
     """
 
     # Default reference used as a dummy during tests.
     default_ref = Reference('branch', 'default', None)
 
     # Methods of vcsserver.hg.HgRemote that are "writing" operations.
     writing_methods = [
         'bookmark',
         'commit',
         'merge',
         'pull',
         'pull_cmd',
         'rebase',
         'strip',
         'tag',
     ]
 
     @pytest.mark.parametrize('method_name, method_args', [
         ('_local_merge', [default_ref, None, None, None, default_ref]),
         ('_local_pull', ['', default_ref]),
         ('bookmark', [None]),
         ('pull', ['', default_ref]),
         ('remove_tag', ['mytag', None]),
         ('strip', [None]),
         ('tag', ['newtag', None]),
     ])
     def test_method_invokes_invalidate_on_remote_repo(
             self, method_name, method_args, backend_hg):
         """
         Check that the listed methods are invalidating the VCSServer cache
         after invoking a writing method of their remote repository object.
         """
         tags = {'mytag': 'mytag-id'}
 
         def add_tag(name, raw_id, *args, **kwds):
             tags[name] = raw_id
 
         repo = backend_hg.repo.scm_instance()
         with patch.object(repo, '_remote') as remote:
             remote.lookup.return_value = ('commit-id', 'commit-idx')
             remote.tags.return_value = tags
             remote._get_tags.return_value = tags
             remote.tag.side_effect = add_tag
 
             # Invoke method.
             method = getattr(repo, method_name)
             method(*method_args)
 
             # Assert that every "writing" method is followed by an invocation
             # of the cache invalidation method.
             for counter, method_call in enumerate(remote.method_calls):
                 call_name = method_call[0]
                 if call_name in self.writing_methods:
                     next_call = remote.method_calls[counter + 1]
                     assert next_call == call.invalidate_vcs_cache()
 
     def _prepare_shadow_repo(self, pull_request):
         """
         Helper that creates a shadow repo that can be used to reproduce the
         CommitDoesNotExistError when pulling in from target and source
         references.
         """
         from rhodecode.model.pull_request import PullRequestModel
 
         target_vcs = pull_request.target_repo.scm_instance()
         target_ref = pull_request.target_ref_parts
         source_ref = pull_request.source_ref_parts
 
         # Create shadow repository.
         pr = PullRequestModel()
         workspace_id = pr._workspace_id(pull_request)
         shadow_repository_path = target_vcs._maybe_prepare_merge_workspace(
-            workspace_id, target_ref)
+            workspace_id, target_ref, source_ref)
         shadow_repo = target_vcs._get_shadow_instance(shadow_repository_path)
 
         # This will populate the cache of the mercurial repository object
         # inside of the VCSServer.
         shadow_repo.get_commit()
 
         return shadow_repo, source_ref, target_ref
 
     @pytest.mark.backends('hg')
     def test_commit_does_not_exist_error_happens(self, pr_util, app):
         """
         This test is somewhat special. It does not really test the system
         instead it is more or less a precondition for the
         "test_commit_does_not_exist_error_does_not_happen". It deactivates the
         cache invalidation and asserts that the error occurs.
         """
         from rhodecode.lib.vcs.exceptions import CommitDoesNotExistError
 
         pull_request = pr_util.create_pull_request()
         target_vcs = pull_request.target_repo.scm_instance()
         source_vcs = pull_request.source_repo.scm_instance()
         shadow_repo, source_ref, target_ref = self._prepare_shadow_repo(
             pull_request)
 
         # Pull from target and source references but without invalidation of
         # RemoteRepo objects and without VCSServer caching of mercurial
         # repository objects.
         with patch.object(shadow_repo._remote, 'invalidate_vcs_cache'):
             # NOTE: Do not use patch.dict() to disable the cache because it
             # restores the WHOLE dict and not only the patched keys.
             shadow_repo._remote._wire['cache'] = False
             shadow_repo._local_pull(target_vcs.path, target_ref)
             shadow_repo._local_pull(source_vcs.path, source_ref)
             shadow_repo._remote._wire.pop('cache')
 
         # Try to lookup the target_ref in shadow repo. This should work because
         # the shadow repo is a clone of the target and always contains all off
         # it's commits in the initial cache.
         shadow_repo.get_commit(target_ref.commit_id)
 
         # If we try to lookup the source_ref it should fail because the shadow
         # repo commit cache doesn't get invalidated. (Due to patched
         # invalidation and caching above).
         with pytest.raises(CommitDoesNotExistError):
             shadow_repo.get_commit(source_ref.commit_id)
 
     @pytest.mark.backends('hg')
     def test_commit_does_not_exist_error_does_not_happen(self, pr_util, app):
         """
         This test simulates a pull request merge in which the pull operations
         are handled by a different VCSServer process than all other operations.
         Without correct cache invalidation this leads to an error when
         retrieving the pulled commits afterwards.
         """
 
         pull_request = pr_util.create_pull_request()
         target_vcs = pull_request.target_repo.scm_instance()
         source_vcs = pull_request.source_repo.scm_instance()
         shadow_repo, source_ref, target_ref = self._prepare_shadow_repo(
             pull_request)
 
         # Pull from target and source references without without VCSServer
         # caching of mercurial repository objects but with active invalidation
         # of RemoteRepo objects.
         # NOTE: Do not use patch.dict() to disable the cache because it
         # restores the WHOLE dict and not only the patched keys.
         shadow_repo._remote._wire['cache'] = False
         shadow_repo._local_pull(target_vcs.path, target_ref)
         shadow_repo._local_pull(source_vcs.path, source_ref)
         shadow_repo._remote._wire.pop('cache')
 
         # Try to lookup the target and source references in shadow repo. This
         # should work because the RemoteRepo object gets invalidated during the
         # above pull operations.
         shadow_repo.get_commit(target_ref.commit_id)
         shadow_repo.get_commit(source_ref.commit_id)