tests: further test fixes

rhodecode/tests/plugin.py

@@ -1,1888 +1,1902 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2019 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 collections
 import datetime
 import hashlib
 import os
 import re
 import pprint
 import shutil
 import socket
 import subprocess32
 import time
 import uuid
 import dateutil.tz
 import functools
 
 import mock
 import pyramid.testing
 import pytest
 import colander
 import requests
 import pyramid.paster
 
 import rhodecode
 from rhodecode.lib.utils2 import AttributeDict
 from rhodecode.model.changeset_status import ChangesetStatusModel
 from rhodecode.model.comment import CommentsModel
 from rhodecode.model.db import (
     PullRequest, Repository, RhodeCodeSetting, ChangesetStatus, RepoGroup,
     UserGroup, RepoRhodeCodeUi, RepoRhodeCodeSetting, RhodeCodeUi)
 from rhodecode.model.meta import Session
 from rhodecode.model.pull_request import PullRequestModel
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.user import UserModel
 from rhodecode.model.settings import VcsSettingsModel
 from rhodecode.model.user_group import UserGroupModel
 from rhodecode.model.integration import IntegrationModel
 from rhodecode.integrations import integration_type_registry
 from rhodecode.integrations.types.base import IntegrationTypeBase
 from rhodecode.lib.utils import repo2db_mapper
 from rhodecode.lib.vcs import create_vcsserver_proxy
 from rhodecode.lib.vcs.backends import get_backend
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.tests import (
     login_user_session, get_new_dir, utils, TESTS_TMP_PATH,
     TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR2_LOGIN,
     TEST_USER_REGULAR_PASS)
 from rhodecode.tests.utils import CustomTestApp, set_anonymous_access
 from rhodecode.tests.fixture import Fixture
 from rhodecode.config import utils as config_utils
 
 def _split_comma(value):
     return value.split(',')
 
 
 def pytest_addoption(parser):
     parser.addoption(
         '--keep-tmp-path', action='store_true',
         help="Keep the test temporary directories")
     parser.addoption(
         '--backends', action='store', type=_split_comma,
         default=['git', 'hg', 'svn'],
         help="Select which backends to test for backend specific tests.")
     parser.addoption(
         '--dbs', action='store', type=_split_comma,
         default=['sqlite'],
         help="Select which database to test for database specific tests. "
              "Possible options are sqlite,postgres,mysql")
     parser.addoption(
         '--appenlight', '--ae', action='store_true',
         help="Track statistics in appenlight.")
     parser.addoption(
         '--appenlight-api-key', '--ae-key',
         help="API key for Appenlight.")
     parser.addoption(
         '--appenlight-url', '--ae-url',
         default="https://ae.rhodecode.com",
         help="Appenlight service URL, defaults to https://ae.rhodecode.com")
     parser.addoption(
         '--sqlite-connection-string', action='store',
         default='', help="Connection string for the dbs tests with SQLite")
     parser.addoption(
         '--postgres-connection-string', action='store',
         default='', help="Connection string for the dbs tests with Postgres")
     parser.addoption(
         '--mysql-connection-string', action='store',
         default='', help="Connection string for the dbs tests with MySQL")
     parser.addoption(
         '--repeat', type=int, default=100,
         help="Number of repetitions in performance tests.")
 
 
 def pytest_configure(config):
     from rhodecode.config import patches
 
 
 def pytest_collection_modifyitems(session, config, items):
     # nottest marked, compare nose, used for transition from nose to pytest
     remaining = [
         i for i in items if getattr(i.obj, '__test__', True)]
     items[:] = remaining
 
 
 def pytest_generate_tests(metafunc):
     # Support test generation based on --backend parameter
     if 'backend_alias' in metafunc.fixturenames:
         backends = get_backends_from_metafunc(metafunc)
         scope = None
         if not backends:
             pytest.skip("Not enabled for any of selected backends")
         metafunc.parametrize('backend_alias', backends, scope=scope)
     elif hasattr(metafunc.function, 'backends'):
         backends = get_backends_from_metafunc(metafunc)
         if not backends:
             pytest.skip("Not enabled for any of selected backends")
 
 
 def get_backends_from_metafunc(metafunc):
     requested_backends = set(metafunc.config.getoption('--backends'))
     if hasattr(metafunc.function, 'backends'):
         # Supported backends by this test function, created from
         # pytest.mark.backends
         backends = metafunc.definition.get_closest_marker('backends').args
     elif hasattr(metafunc.cls, 'backend_alias'):
         # Support class attribute "backend_alias", this is mainly
         # for legacy reasons for tests not yet using pytest.mark.backends
         backends = [metafunc.cls.backend_alias]
     else:
         backends = metafunc.config.getoption('--backends')
     return requested_backends.intersection(backends)
 
 
 @pytest.fixture(scope='session', autouse=True)
 def activate_example_rcextensions(request):
     """
     Patch in an example rcextensions module which verifies passed in kwargs.
     """
     from rhodecode.config import rcextensions
 
     old_extensions = rhodecode.EXTENSIONS
     rhodecode.EXTENSIONS = rcextensions
     rhodecode.EXTENSIONS.calls = collections.defaultdict(list)
 
     @request.addfinalizer
     def cleanup():
         rhodecode.EXTENSIONS = old_extensions
 
 
 @pytest.fixture
 def capture_rcextensions():
     """
     Returns the recorded calls to entry points in rcextensions.
     """
     calls = rhodecode.EXTENSIONS.calls
     calls.clear()
     # Note: At this moment, it is still the empty dict, but that will
     # be filled during the test run and since it is a reference this
     # is enough to make it work.
     return calls
 
 
 @pytest.fixture(scope='session')
 def http_environ_session():
     """
     Allow to use "http_environ" in session scope.
     """
     return plain_http_environ()
 
 
 def plain_http_host_stub():
     """
     Value of HTTP_HOST in the test run.
     """
     return 'example.com:80'
 
 
 @pytest.fixture
 def http_host_stub():
     """
     Value of HTTP_HOST in the test run.
     """
     return plain_http_host_stub()
 
 
 def plain_http_host_only_stub():
     """
     Value of HTTP_HOST in the test run.
     """
     return plain_http_host_stub().split(':')[0]
 
 
 @pytest.fixture
 def http_host_only_stub():
     """
     Value of HTTP_HOST in the test run.
     """
     return plain_http_host_only_stub()
 
 
 def plain_http_environ():
     """
     HTTP extra environ keys.
 
     User by the test application and as well for setting up the pylons
     environment. In the case of the fixture "app" it should be possible
     to override this for a specific test case.
     """
     return {
         'SERVER_NAME': plain_http_host_only_stub(),
         'SERVER_PORT': plain_http_host_stub().split(':')[1],
         'HTTP_HOST': plain_http_host_stub(),
         'HTTP_USER_AGENT': 'rc-test-agent',
         'REQUEST_METHOD': 'GET'
     }
 
 
 @pytest.fixture
 def http_environ():
     """
     HTTP extra environ keys.
 
     User by the test application and as well for setting up the pylons
     environment. In the case of the fixture "app" it should be possible
     to override this for a specific test case.
     """
     return plain_http_environ()
 
 
 @pytest.fixture(scope='session')
 def baseapp(ini_config, vcsserver, http_environ_session):
     from rhodecode.lib.pyramid_utils import get_app_config
     from rhodecode.config.middleware import make_pyramid_app
 
     print("Using the RhodeCode configuration:{}".format(ini_config))
     pyramid.paster.setup_logging(ini_config)
 
     settings = get_app_config(ini_config)
     app = make_pyramid_app({'__file__': ini_config}, **settings)
 
     return app
 
 
 @pytest.fixture(scope='function')
 def app(request, config_stub, baseapp, http_environ):
     app = CustomTestApp(
         baseapp,
         extra_environ=http_environ)
     if request.cls:
         request.cls.app = app
     return app
 
 
 @pytest.fixture(scope='session')
 def app_settings(baseapp, ini_config):
     """
     Settings dictionary used to create the app.
 
     Parses the ini file and passes the result through the sanitize and apply
     defaults mechanism in `rhodecode.config.middleware`.
     """
     return baseapp.config.get_settings()
 
 
 @pytest.fixture(scope='session')
 def db_connection(ini_settings):
     # Initialize the database connection.
     config_utils.initialize_database(ini_settings)
 
 
 LoginData = collections.namedtuple('LoginData', ('csrf_token', 'user'))
 
 
 def _autologin_user(app, *args):
     session = login_user_session(app, *args)
     csrf_token = rhodecode.lib.auth.get_csrf_token(session)
     return LoginData(csrf_token, session['rhodecode_user'])
 
 
 @pytest.fixture
 def autologin_user(app):
     """
     Utility fixture which makes sure that the admin user is logged in
     """
     return _autologin_user(app)
 
 
 @pytest.fixture
 def autologin_regular_user(app):
     """
     Utility fixture which makes sure that the regular user is logged in
     """
     return _autologin_user(
         app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
 
 
 @pytest.fixture(scope='function')
 def csrf_token(request, autologin_user):
     return autologin_user.csrf_token
 
 
 @pytest.fixture(scope='function')
 def xhr_header(request):
     return {'HTTP_X_REQUESTED_WITH': 'XMLHttpRequest'}
 
 
 @pytest.fixture
 def real_crypto_backend(monkeypatch):
     """
     Switch the production crypto backend on for this test.
 
     During the test run the crypto backend is replaced with a faster
     implementation based on the MD5 algorithm.
     """
     monkeypatch.setattr(rhodecode, 'is_test', False)
 
 
 @pytest.fixture(scope='class')
 def index_location(request, baseapp):
     index_location = baseapp.config.get_settings()['search.location']
     if request.cls:
         request.cls.index_location = index_location
     return index_location
 
 
 @pytest.fixture(scope='session', autouse=True)
 def tests_tmp_path(request):
     """
     Create temporary directory to be used during the test session.
     """
     if not os.path.exists(TESTS_TMP_PATH):
         os.makedirs(TESTS_TMP_PATH)
 
     if not request.config.getoption('--keep-tmp-path'):
         @request.addfinalizer
         def remove_tmp_path():
             shutil.rmtree(TESTS_TMP_PATH)
 
     return TESTS_TMP_PATH
 
 
 @pytest.fixture
 def test_repo_group(request):
     """
     Create a temporary repository group, and destroy it after
     usage automatically
     """
     fixture = Fixture()
     repogroupid = 'test_repo_group_%s' % str(time.time()).replace('.', '')
     repo_group = fixture.create_repo_group(repogroupid)
 
     def _cleanup():
         fixture.destroy_repo_group(repogroupid)
 
     request.addfinalizer(_cleanup)
     return repo_group
 
 
 @pytest.fixture
 def test_user_group(request):
     """
     Create a temporary user group, and destroy it after
     usage automatically
     """
     fixture = Fixture()
     usergroupid = 'test_user_group_%s' % str(time.time()).replace('.', '')
     user_group = fixture.create_user_group(usergroupid)
 
     def _cleanup():
         fixture.destroy_user_group(user_group)
 
     request.addfinalizer(_cleanup)
     return user_group
 
 
 @pytest.fixture(scope='session')
 def test_repo(request):
     container = TestRepoContainer()
     request.addfinalizer(container._cleanup)
     return container
 
 
 class TestRepoContainer(object):
     """
     Container for test repositories which are used read only.
 
     Repositories will be created on demand and re-used during the lifetime
     of this object.
 
     Usage to get the svn test repository "minimal"::
 
        test_repo = TestContainer()
        repo = test_repo('minimal', 'svn')
 
     """
 
     dump_extractors = {
         'git': utils.extract_git_repo_from_dump,
         'hg': utils.extract_hg_repo_from_dump,
         'svn': utils.extract_svn_repo_from_dump,
     }
 
     def __init__(self):
         self._cleanup_repos = []
         self._fixture = Fixture()
         self._repos = {}
 
     def __call__(self, dump_name, backend_alias, config=None):
         key = (dump_name, backend_alias)
         if key not in self._repos:
             repo = self._create_repo(dump_name, backend_alias, config)
             self._repos[key] = repo.repo_id
         return Repository.get(self._repos[key])
 
     def _create_repo(self, dump_name, backend_alias, config):
         repo_name = '%s-%s' % (backend_alias, dump_name)
         backend_class = get_backend(backend_alias)
         dump_extractor = self.dump_extractors[backend_alias]
         repo_path = dump_extractor(dump_name, repo_name)
 
         vcs_repo = backend_class(repo_path, config=config)
         repo2db_mapper({repo_name: vcs_repo})
 
         repo = RepoModel().get_by_repo_name(repo_name)
         self._cleanup_repos.append(repo_name)
         return repo
 
     def _cleanup(self):
         for repo_name in reversed(self._cleanup_repos):
             self._fixture.destroy_repo(repo_name)
 
 
 def backend_base(request, backend_alias, baseapp, test_repo):
     if backend_alias not in request.config.getoption('--backends'):
         pytest.skip("Backend %s not selected." % (backend_alias, ))
 
     utils.check_xfail_backends(request.node, backend_alias)
     utils.check_skip_backends(request.node, backend_alias)
 
     repo_name = 'vcs_test_%s' % (backend_alias, )
     backend = Backend(
         alias=backend_alias,
         repo_name=repo_name,
         test_name=request.node.name,
         test_repo_container=test_repo)
     request.addfinalizer(backend.cleanup)
     return backend
 
 
 @pytest.fixture
 def backend(request, backend_alias, baseapp, test_repo):
     """
     Parametrized fixture which represents a single backend implementation.
 
     It respects the option `--backends` to focus the test run on specific
     backend implementations.
 
     It also supports `pytest.mark.xfail_backends` to mark tests as failing
     for specific backends. This is intended as a utility for incremental
     development of a new backend implementation.
     """
     return backend_base(request, backend_alias, baseapp, test_repo)
 
 
 @pytest.fixture
 def backend_git(request, baseapp, test_repo):
     return backend_base(request, 'git', baseapp, test_repo)
 
 
 @pytest.fixture
 def backend_hg(request, baseapp, test_repo):
     return backend_base(request, 'hg', baseapp, test_repo)
 
 
 @pytest.fixture
 def backend_svn(request, baseapp, test_repo):
     return backend_base(request, 'svn', baseapp, test_repo)
 
 
 @pytest.fixture
 def backend_random(backend_git):
     """
     Use this to express that your tests need "a backend.
 
     A few of our tests need a backend, so that we can run the code. This
     fixture is intended to be used for such cases. It will pick one of the
     backends and run the tests.
 
     The fixture `backend` would run the test multiple times for each
     available backend which is a pure waste of time if the test is
     independent of the backend type.
     """
     # TODO: johbo: Change this to pick a random backend
     return backend_git
 
 
 @pytest.fixture
 def backend_stub(backend_git):
     """
     Use this to express that your tests need a backend stub
 
     TODO: mikhail: Implement a real stub logic instead of returning
     a git backend
     """
     return backend_git
 
 
 @pytest.fixture
 def repo_stub(backend_stub):
     """
     Use this to express that your tests need a repository stub
     """
     return backend_stub.create_repo()
 
 
 class Backend(object):
     """
     Represents the test configuration for one supported backend
 
     Provides easy access to different test repositories based on
     `__getitem__`. Such repositories will only be created once per test
     session.
     """
 
     invalid_repo_name = re.compile(r'[^0-9a-zA-Z]+')
     _master_repo = None
     _commit_ids = {}
 
     def __init__(self, alias, repo_name, test_name, test_repo_container):
         self.alias = alias
         self.repo_name = repo_name
         self._cleanup_repos = []
         self._test_name = test_name
         self._test_repo_container = test_repo_container
         # TODO: johbo: Used as a delegate interim. Not yet sure if Backend or
         # Fixture will survive in the end.
         self._fixture = Fixture()
 
     def __getitem__(self, key):
         return self._test_repo_container(key, self.alias)
 
     def create_test_repo(self, key, config=None):
         return self._test_repo_container(key, self.alias, config)
 
     @property
     def repo(self):
         """
         Returns the "current" repository. This is the vcs_test repo or the
         last repo which has been created with `create_repo`.
         """
         from rhodecode.model.db import Repository
         return Repository.get_by_repo_name(self.repo_name)
 
     @property
     def default_branch_name(self):
         VcsRepository = get_backend(self.alias)
         return VcsRepository.DEFAULT_BRANCH_NAME
 
     @property
     def default_head_id(self):
         """
         Returns the default head id of the underlying backend.
 
         This will be the default branch name in case the backend does have a
         default branch. In the other cases it will point to a valid head
         which can serve as the base to create a new commit on top of it.
         """
         vcsrepo = self.repo.scm_instance()
         head_id = (
             vcsrepo.DEFAULT_BRANCH_NAME or
             vcsrepo.commit_ids[-1])
         return head_id
 
     @property
     def commit_ids(self):
         """
         Returns the list of commits for the last created repository
         """
         return self._commit_ids
 
     def create_master_repo(self, commits):
         """
         Create a repository and remember it as a template.
 
         This allows to easily create derived repositories to construct
         more complex scenarios for diff, compare and pull requests.
 
         Returns a commit map which maps from commit message to raw_id.
         """
         self._master_repo = self.create_repo(commits=commits)
         return self._commit_ids
 
     def create_repo(
             self, commits=None, number_of_commits=0, heads=None,
             name_suffix=u'', bare=False, **kwargs):
         """
         Create a repository and record it for later cleanup.
 
         :param commits: Optional. A sequence of dict instances.
            Will add a commit per entry to the new repository.
         :param number_of_commits: Optional. If set to a number, this number of
            commits will be added to the new repository.
         :param heads: Optional. Can be set to a sequence of of commit
            names which shall be pulled in from the master repository.
         :param name_suffix: adds special suffix to generated repo name
         :param bare: set a repo as bare (no checkout)
         """
         self.repo_name = self._next_repo_name() + name_suffix
         repo = self._fixture.create_repo(
             self.repo_name, repo_type=self.alias, bare=bare, **kwargs)
         self._cleanup_repos.append(repo.repo_name)
 
         commits = commits or [
             {'message': 'Commit %s of %s' % (x, self.repo_name)}
             for x in range(number_of_commits)]
         vcs_repo = repo.scm_instance()
         vcs_repo.count()
         self._add_commits_to_repo(vcs_repo, commits)
         if heads:
             self.pull_heads(repo, heads)
 
         return repo
 
     def pull_heads(self, repo, heads):
         """
         Make sure that repo contains all commits mentioned in `heads`
         """
         vcsmaster = self._master_repo.scm_instance()
         vcsrepo = repo.scm_instance()
         vcsrepo.config.clear_section('hooks')
         commit_ids = [self._commit_ids[h] for h in heads]
         vcsrepo.pull(vcsmaster.path, commit_ids=commit_ids)
 
     def create_fork(self):
         repo_to_fork = self.repo_name
         self.repo_name = self._next_repo_name()
         repo = self._fixture.create_fork(repo_to_fork, self.repo_name)
         self._cleanup_repos.append(self.repo_name)
         return repo
 
     def new_repo_name(self, suffix=u''):
         self.repo_name = self._next_repo_name() + suffix
         self._cleanup_repos.append(self.repo_name)
         return self.repo_name
 
     def _next_repo_name(self):
         return u"%s_%s" % (
             self.invalid_repo_name.sub(u'_', self._test_name), len(self._cleanup_repos))
 
     def ensure_file(self, filename, content='Test content\n'):
         assert self._cleanup_repos, "Avoid writing into vcs_test repos"
         commits = [
             {'added': [
                 FileNode(filename, content=content),
             ]},
         ]
         self._add_commits_to_repo(self.repo.scm_instance(), commits)
 
     def enable_downloads(self):
         repo = self.repo
         repo.enable_downloads = True
         Session().add(repo)
         Session().commit()
 
     def cleanup(self):
         for repo_name in reversed(self._cleanup_repos):
             self._fixture.destroy_repo(repo_name)
 
     def _add_commits_to_repo(self, repo, commits):
         commit_ids = _add_commits_to_repo(repo, commits)
         if not commit_ids:
             return
         self._commit_ids = commit_ids
 
         # Creating refs for Git to allow fetching them from remote repository
         if self.alias == 'git':
             refs = {}
             for message in self._commit_ids:
                 # TODO: mikhail: do more special chars replacements
                 ref_name = 'refs/test-refs/{}'.format(
                     message.replace(' ', ''))
                 refs[ref_name] = self._commit_ids[message]
             self._create_refs(repo, refs)
 
     def _create_refs(self, repo, refs):
         for ref_name in refs:
             repo.set_refs(ref_name, refs[ref_name])
 
 
 def vcsbackend_base(request, backend_alias, tests_tmp_path, baseapp, test_repo):
     if backend_alias not in request.config.getoption('--backends'):
         pytest.skip("Backend %s not selected." % (backend_alias, ))
 
     utils.check_xfail_backends(request.node, backend_alias)
     utils.check_skip_backends(request.node, backend_alias)
 
     repo_name = 'vcs_test_%s' % (backend_alias, )
     repo_path = os.path.join(tests_tmp_path, repo_name)
     backend = VcsBackend(
         alias=backend_alias,
         repo_path=repo_path,
         test_name=request.node.name,
         test_repo_container=test_repo)
     request.addfinalizer(backend.cleanup)
     return backend
 
 
 @pytest.fixture
 def vcsbackend(request, backend_alias, tests_tmp_path, baseapp, test_repo):
     """
     Parametrized fixture which represents a single vcs backend implementation.
 
     See the fixture `backend` for more details. This one implements the same
     concept, but on vcs level. So it does not provide model instances etc.
 
     Parameters are generated dynamically, see :func:`pytest_generate_tests`
     for how this works.
     """
     return vcsbackend_base(request, backend_alias, tests_tmp_path, baseapp, test_repo)
 
 
 @pytest.fixture
 def vcsbackend_git(request, tests_tmp_path, baseapp, test_repo):
     return vcsbackend_base(request, 'git', tests_tmp_path, baseapp, test_repo)
 
 
 @pytest.fixture
 def vcsbackend_hg(request, tests_tmp_path, baseapp, test_repo):
     return vcsbackend_base(request, 'hg', tests_tmp_path, baseapp, test_repo)
 
 
 @pytest.fixture
 def vcsbackend_svn(request, tests_tmp_path, baseapp, test_repo):
     return vcsbackend_base(request, 'svn', tests_tmp_path, baseapp, test_repo)
 
 
 @pytest.fixture
 def vcsbackend_stub(vcsbackend_git):
     """
     Use this to express that your test just needs a stub of a vcsbackend.
 
     Plan is to eventually implement an in-memory stub to speed tests up.
     """
     return vcsbackend_git
 
 
 class VcsBackend(object):
     """
     Represents the test configuration for one supported vcs backend.
     """
 
     invalid_repo_name = re.compile(r'[^0-9a-zA-Z]+')
 
     def __init__(self, alias, repo_path, test_name, test_repo_container):
         self.alias = alias
         self._repo_path = repo_path
         self._cleanup_repos = []
         self._test_name = test_name
         self._test_repo_container = test_repo_container
 
     def __getitem__(self, key):
         return self._test_repo_container(key, self.alias).scm_instance()
 
     @property
     def repo(self):
         """
         Returns the "current" repository. This is the vcs_test repo of the last
         repo which has been created.
         """
         Repository = get_backend(self.alias)
         return Repository(self._repo_path)
 
     @property
     def backend(self):
         """
         Returns the backend implementation class.
         """
         return get_backend(self.alias)
 
     def create_repo(self, commits=None, number_of_commits=0, _clone_repo=None,
                     bare=False):
         repo_name = self._next_repo_name()
         self._repo_path = get_new_dir(repo_name)
         repo_class = get_backend(self.alias)
         src_url = None
         if _clone_repo:
             src_url = _clone_repo.path
         repo = repo_class(self._repo_path, create=True, src_url=src_url, bare=bare)
         self._cleanup_repos.append(repo)
 
         commits = commits or [
             {'message': 'Commit %s of %s' % (x, repo_name)}
             for x in xrange(number_of_commits)]
         _add_commits_to_repo(repo, commits)
         return repo
 
     def clone_repo(self, repo):
         return self.create_repo(_clone_repo=repo)
 
     def cleanup(self):
         for repo in self._cleanup_repos:
             shutil.rmtree(repo.path)
 
     def new_repo_path(self):
         repo_name = self._next_repo_name()
         self._repo_path = get_new_dir(repo_name)
         return self._repo_path
 
     def _next_repo_name(self):
         return "%s_%s" % (
             self.invalid_repo_name.sub('_', self._test_name),
             len(self._cleanup_repos))
 
     def add_file(self, repo, filename, content='Test content\n'):
         imc = repo.in_memory_commit
         imc.add(FileNode(filename, content=content))
         imc.commit(
             message=u'Automatic commit from vcsbackend fixture',
             author=u'Automatic')
 
     def ensure_file(self, filename, content='Test content\n'):
         assert self._cleanup_repos, "Avoid writing into vcs_test repos"
         self.add_file(self.repo, filename, content)
 
 
 def _add_commits_to_repo(vcs_repo, commits):
     commit_ids = {}
     if not commits:
         return commit_ids
 
     imc = vcs_repo.in_memory_commit
     commit = None
 
     for idx, commit in enumerate(commits):
         message = unicode(commit.get('message', 'Commit %s' % idx))
 
         for node in commit.get('added', []):
             imc.add(FileNode(node.path, content=node.content))
         for node in commit.get('changed', []):
             imc.change(FileNode(node.path, content=node.content))
         for node in commit.get('removed', []):
             imc.remove(FileNode(node.path))
 
         parents = [
             vcs_repo.get_commit(commit_id=commit_ids[p])
             for p in commit.get('parents', [])]
 
         operations = ('added', 'changed', 'removed')
         if not any((commit.get(o) for o in operations)):
             imc.add(FileNode('file_%s' % idx, content=message))
 
         commit = imc.commit(
             message=message,
             author=unicode(commit.get('author', 'Automatic')),
             date=commit.get('date'),
             branch=commit.get('branch'),
             parents=parents)
 
         commit_ids[commit.message] = commit.raw_id
 
     return commit_ids
 
 
 @pytest.fixture
 def reposerver(request):
     """
     Allows to serve a backend repository
     """
 
     repo_server = RepoServer()
     request.addfinalizer(repo_server.cleanup)
     return repo_server
 
 
 class RepoServer(object):
     """
     Utility to serve a local repository for the duration of a test case.
 
     Supports only Subversion so far.
     """
 
     url = None
 
     def __init__(self):
         self._cleanup_servers = []
 
     def serve(self, vcsrepo):
         if vcsrepo.alias != 'svn':
             raise TypeError("Backend %s not supported" % vcsrepo.alias)
 
         proc = subprocess32.Popen(
             ['svnserve', '-d', '--foreground', '--listen-host', 'localhost',
              '--root', vcsrepo.path])
         self._cleanup_servers.append(proc)
         self.url = 'svn://localhost'
 
     def cleanup(self):
         for proc in self._cleanup_servers:
             proc.terminate()
 
 
 @pytest.fixture
 def pr_util(backend, request, config_stub):
     """
     Utility for tests of models and for functional tests around pull requests.
 
     It gives an instance of :class:`PRTestUtility` which provides various
     utility methods around one pull request.
 
     This fixture uses `backend` and inherits its parameterization.
     """
 
     util = PRTestUtility(backend)
     request.addfinalizer(util.cleanup)
 
     return util
 
 
 class PRTestUtility(object):
 
     pull_request = None
     pull_request_id = None
     mergeable_patcher = None
     mergeable_mock = None
     notification_patcher = None
 
     def __init__(self, backend):
         self.backend = backend
 
     def create_pull_request(
             self, commits=None, target_head=None, source_head=None,
             revisions=None, approved=False, author=None, mergeable=False,
             enable_notifications=True, name_suffix=u'', reviewers=None,
             title=u"Test", description=u"Description"):
         self.set_mergeable(mergeable)
         if not enable_notifications:
             # mock notification side effect
             self.notification_patcher = mock.patch(
                 'rhodecode.model.notification.NotificationModel.create')
             self.notification_patcher.start()
 
         if not self.pull_request:
             if not commits:
                 commits = [
                     {'message': 'c1'},
                     {'message': 'c2'},
                     {'message': 'c3'},
                 ]
                 target_head = 'c1'
                 source_head = 'c2'
                 revisions = ['c2']
 
             self.commit_ids = self.backend.create_master_repo(commits)
             self.target_repository = self.backend.create_repo(
                 heads=[target_head], name_suffix=name_suffix)
             self.source_repository = self.backend.create_repo(
                 heads=[source_head], name_suffix=name_suffix)
             self.author = author or UserModel().get_by_username(
                 TEST_USER_ADMIN_LOGIN)
 
             model = PullRequestModel()
             self.create_parameters = {
                 'created_by': self.author,
                 'source_repo': self.source_repository.repo_name,
                 'source_ref': self._default_branch_reference(source_head),
                 'target_repo': self.target_repository.repo_name,
                 'target_ref': self._default_branch_reference(target_head),
                 'revisions': [self.commit_ids[r] for r in revisions],
                 'reviewers': reviewers or self._get_reviewers(),
                 'title': title,
                 'description': description,
             }
             self.pull_request = model.create(**self.create_parameters)
             assert model.get_versions(self.pull_request) == []
 
             self.pull_request_id = self.pull_request.pull_request_id
 
             if approved:
                 self.approve()
 
             Session().add(self.pull_request)
             Session().commit()
 
         return self.pull_request
 
     def approve(self):
         self.create_status_votes(
             ChangesetStatus.STATUS_APPROVED,
             *self.pull_request.reviewers)
 
     def close(self):
         PullRequestModel().close_pull_request(self.pull_request, self.author)
 
     def _default_branch_reference(self, commit_message):
         reference = '%s:%s:%s' % (
             'branch',
             self.backend.default_branch_name,
             self.commit_ids[commit_message])
         return reference
 
     def _get_reviewers(self):
         return [
             (TEST_USER_REGULAR_LOGIN, ['default1'], False, []),
             (TEST_USER_REGULAR2_LOGIN, ['default2'], False, []),
         ]
 
     def update_source_repository(self, head=None):
         heads = [head or 'c3']
         self.backend.pull_heads(self.source_repository, heads=heads)
 
     def add_one_commit(self, head=None):
         self.update_source_repository(head=head)
         old_commit_ids = set(self.pull_request.revisions)
         PullRequestModel().update_commits(self.pull_request)
         commit_ids = set(self.pull_request.revisions)
         new_commit_ids = commit_ids - old_commit_ids
         assert len(new_commit_ids) == 1
         return new_commit_ids.pop()
 
     def remove_one_commit(self):
         assert len(self.pull_request.revisions) == 2
         source_vcs = self.source_repository.scm_instance()
         removed_commit_id = source_vcs.commit_ids[-1]
 
         # TODO: johbo: Git and Mercurial have an inconsistent vcs api here,
         # remove the if once that's sorted out.
         if self.backend.alias == "git":
             kwargs = {'branch_name': self.backend.default_branch_name}
         else:
             kwargs = {}
         source_vcs.strip(removed_commit_id, **kwargs)
 
         PullRequestModel().update_commits(self.pull_request)
         assert len(self.pull_request.revisions) == 1
         return removed_commit_id
 
     def create_comment(self, linked_to=None):
         comment = CommentsModel().create(
             text=u"Test comment",
             repo=self.target_repository.repo_name,
             user=self.author,
             pull_request=self.pull_request)
         assert comment.pull_request_version_id is None
 
         if linked_to:
             PullRequestModel()._link_comments_to_version(linked_to)
 
         return comment
 
     def create_inline_comment(
             self, linked_to=None, line_no=u'n1', file_path='file_1'):
         comment = CommentsModel().create(
             text=u"Test comment",
             repo=self.target_repository.repo_name,
             user=self.author,
             line_no=line_no,
             f_path=file_path,
             pull_request=self.pull_request)
         assert comment.pull_request_version_id is None
 
         if linked_to:
             PullRequestModel()._link_comments_to_version(linked_to)
 
         return comment
 
     def create_version_of_pull_request(self):
         pull_request = self.create_pull_request()
         version = PullRequestModel()._create_version_from_snapshot(
             pull_request)
         return version
 
     def create_status_votes(self, status, *reviewers):
         for reviewer in reviewers:
             ChangesetStatusModel().set_status(
                 repo=self.pull_request.target_repo,
                 status=status,
                 user=reviewer.user_id,
                 pull_request=self.pull_request)
 
     def set_mergeable(self, value):
         if not self.mergeable_patcher:
             self.mergeable_patcher = mock.patch.object(
                 VcsSettingsModel, 'get_general_settings')
             self.mergeable_mock = self.mergeable_patcher.start()
         self.mergeable_mock.return_value = {
             'rhodecode_pr_merge_enabled': value}
 
     def cleanup(self):
         # In case the source repository is already cleaned up, the pull
         # request will already be deleted.
         pull_request = PullRequest().get(self.pull_request_id)
         if pull_request:
             PullRequestModel().delete(pull_request, pull_request.author)
             Session().commit()
 
         if self.notification_patcher:
             self.notification_patcher.stop()
 
         if self.mergeable_patcher:
             self.mergeable_patcher.stop()
 
 
 @pytest.fixture
 def user_admin(baseapp):
     """
     Provides the default admin test user as an instance of `db.User`.
     """
     user = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
     return user
 
 
 @pytest.fixture
 def user_regular(baseapp):
     """
     Provides the default regular test user as an instance of `db.User`.
     """
     user = UserModel().get_by_username(TEST_USER_REGULAR_LOGIN)
     return user
 
 
 @pytest.fixture
 def user_util(request, db_connection):
     """
     Provides a wired instance of `UserUtility` with integrated cleanup.
     """
     utility = UserUtility(test_name=request.node.name)
     request.addfinalizer(utility.cleanup)
     return utility
 
 
 # TODO: johbo: Split this up into utilities per domain or something similar
 class UserUtility(object):
 
     def __init__(self, test_name="test"):
         self._test_name = self._sanitize_name(test_name)
         self.fixture = Fixture()
         self.repo_group_ids = []
         self.repos_ids = []
         self.user_ids = []
         self.user_group_ids = []
         self.user_repo_permission_ids = []
         self.user_group_repo_permission_ids = []
         self.user_repo_group_permission_ids = []
         self.user_group_repo_group_permission_ids = []
         self.user_user_group_permission_ids = []
         self.user_group_user_group_permission_ids = []
         self.user_permissions = []
 
     def _sanitize_name(self, name):
         for char in ['[', ']']:
             name = name.replace(char, '_')
         return name
 
     def create_repo_group(
             self, owner=TEST_USER_ADMIN_LOGIN, auto_cleanup=True):
         group_name = "{prefix}_repogroup_{count}".format(
             prefix=self._test_name,
             count=len(self.repo_group_ids))
         repo_group = self.fixture.create_repo_group(
             group_name, cur_user=owner)
         if auto_cleanup:
             self.repo_group_ids.append(repo_group.group_id)
         return repo_group
 
     def create_repo(self, owner=TEST_USER_ADMIN_LOGIN, parent=None,
                     auto_cleanup=True, repo_type='hg', bare=False):
         repo_name = "{prefix}_repository_{count}".format(
             prefix=self._test_name,
             count=len(self.repos_ids))
 
         repository = self.fixture.create_repo(
             repo_name, cur_user=owner, repo_group=parent, repo_type=repo_type, bare=bare)
         if auto_cleanup:
             self.repos_ids.append(repository.repo_id)
         return repository
 
     def create_user(self, auto_cleanup=True, **kwargs):
         user_name = "{prefix}_user_{count}".format(
             prefix=self._test_name,
             count=len(self.user_ids))
         user = self.fixture.create_user(user_name, **kwargs)
         if auto_cleanup:
             self.user_ids.append(user.user_id)
         return user
 
     def create_additional_user_email(self, user, email):
         uem = self.fixture.create_additional_user_email(user=user, email=email)
         return uem
 
     def create_user_with_group(self):
         user = self.create_user()
         user_group = self.create_user_group(members=[user])
         return user, user_group
 
     def create_user_group(self, owner=TEST_USER_ADMIN_LOGIN, members=None,
                           auto_cleanup=True, **kwargs):
         group_name = "{prefix}_usergroup_{count}".format(
             prefix=self._test_name,
             count=len(self.user_group_ids))
         user_group = self.fixture.create_user_group(
             group_name, cur_user=owner, **kwargs)
 
         if auto_cleanup:
             self.user_group_ids.append(user_group.users_group_id)
         if members:
             for user in members:
                 UserGroupModel().add_user_to_group(user_group, user)
         return user_group
 
     def grant_user_permission(self, user_name, permission_name):
         self.inherit_default_user_permissions(user_name, False)
         self.user_permissions.append((user_name, permission_name))
 
     def grant_user_permission_to_repo_group(
             self, repo_group, user, permission_name):
         permission = RepoGroupModel().grant_user_permission(
             repo_group, user, permission_name)
         self.user_repo_group_permission_ids.append(
             (repo_group.group_id, user.user_id))
         return permission
 
     def grant_user_group_permission_to_repo_group(
             self, repo_group, user_group, permission_name):
         permission = RepoGroupModel().grant_user_group_permission(
             repo_group, user_group, permission_name)
         self.user_group_repo_group_permission_ids.append(
             (repo_group.group_id, user_group.users_group_id))
         return permission
 
     def grant_user_permission_to_repo(
             self, repo, user, permission_name):
         permission = RepoModel().grant_user_permission(
             repo, user, permission_name)
         self.user_repo_permission_ids.append(
             (repo.repo_id, user.user_id))
         return permission
 
     def grant_user_group_permission_to_repo(
             self, repo, user_group, permission_name):
         permission = RepoModel().grant_user_group_permission(
             repo, user_group, permission_name)
         self.user_group_repo_permission_ids.append(
             (repo.repo_id, user_group.users_group_id))
         return permission
 
     def grant_user_permission_to_user_group(
             self, target_user_group, user, permission_name):
         permission = UserGroupModel().grant_user_permission(
             target_user_group, user, permission_name)
         self.user_user_group_permission_ids.append(
             (target_user_group.users_group_id, user.user_id))
         return permission
 
     def grant_user_group_permission_to_user_group(
             self, target_user_group, user_group, permission_name):
         permission = UserGroupModel().grant_user_group_permission(
             target_user_group, user_group, permission_name)
         self.user_group_user_group_permission_ids.append(
             (target_user_group.users_group_id, user_group.users_group_id))
         return permission
 
     def revoke_user_permission(self, user_name, permission_name):
         self.inherit_default_user_permissions(user_name, True)
         UserModel().revoke_perm(user_name, permission_name)
 
     def inherit_default_user_permissions(self, user_name, value):
         user = UserModel().get_by_username(user_name)
         user.inherit_default_permissions = value
         Session().add(user)
         Session().commit()
 
     def cleanup(self):
         self._cleanup_permissions()
         self._cleanup_repos()
         self._cleanup_repo_groups()
         self._cleanup_user_groups()
         self._cleanup_users()
 
     def _cleanup_permissions(self):
         if self.user_permissions:
             for user_name, permission_name in self.user_permissions:
                 self.revoke_user_permission(user_name, permission_name)
 
         for permission in self.user_repo_permission_ids:
             RepoModel().revoke_user_permission(*permission)
 
         for permission in self.user_group_repo_permission_ids:
             RepoModel().revoke_user_group_permission(*permission)
 
         for permission in self.user_repo_group_permission_ids:
             RepoGroupModel().revoke_user_permission(*permission)
 
         for permission in self.user_group_repo_group_permission_ids:
             RepoGroupModel().revoke_user_group_permission(*permission)
 
         for permission in self.user_user_group_permission_ids:
             UserGroupModel().revoke_user_permission(*permission)
 
         for permission in self.user_group_user_group_permission_ids:
             UserGroupModel().revoke_user_group_permission(*permission)
 
     def _cleanup_repo_groups(self):
         def _repo_group_compare(first_group_id, second_group_id):
             """
             Gives higher priority to the groups with the most complex paths
             """
             first_group = RepoGroup.get(first_group_id)
             second_group = RepoGroup.get(second_group_id)
             first_group_parts = (
                 len(first_group.group_name.split('/')) if first_group else 0)
             second_group_parts = (
                 len(second_group.group_name.split('/')) if second_group else 0)
             return cmp(second_group_parts, first_group_parts)
 
         sorted_repo_group_ids = sorted(
             self.repo_group_ids, cmp=_repo_group_compare)
         for repo_group_id in sorted_repo_group_ids:
             self.fixture.destroy_repo_group(repo_group_id)
 
     def _cleanup_repos(self):
         sorted_repos_ids = sorted(self.repos_ids)
         for repo_id in sorted_repos_ids:
             self.fixture.destroy_repo(repo_id)
 
     def _cleanup_user_groups(self):
         def _user_group_compare(first_group_id, second_group_id):
             """
             Gives higher priority to the groups with the most complex paths
             """
             first_group = UserGroup.get(first_group_id)
             second_group = UserGroup.get(second_group_id)
             first_group_parts = (
                 len(first_group.users_group_name.split('/'))
                 if first_group else 0)
             second_group_parts = (
                 len(second_group.users_group_name.split('/'))
                 if second_group else 0)
             return cmp(second_group_parts, first_group_parts)
 
         sorted_user_group_ids = sorted(
             self.user_group_ids, cmp=_user_group_compare)
         for user_group_id in sorted_user_group_ids:
             self.fixture.destroy_user_group(user_group_id)
 
     def _cleanup_users(self):
         for user_id in self.user_ids:
             self.fixture.destroy_user(user_id)
 
 
 # TODO: Think about moving this into a pytest-pyro package and make it a
 # pytest plugin
 @pytest.hookimpl(tryfirst=True, hookwrapper=True)
 def pytest_runtest_makereport(item, call):
     """
     Adding the remote traceback if the exception has this information.
 
     VCSServer attaches this information as the attribute `_vcs_server_traceback`
     to the exception instance.
     """
     outcome = yield
     report = outcome.get_result()
     if call.excinfo:
         _add_vcsserver_remote_traceback(report, call.excinfo.value)
 
 
 def _add_vcsserver_remote_traceback(report, exc):
     vcsserver_traceback = getattr(exc, '_vcs_server_traceback', None)
 
     if vcsserver_traceback:
         section = 'VCSServer remote traceback ' + report.when
         report.sections.append((section, vcsserver_traceback))
 
 
 @pytest.fixture(scope='session')
 def testrun():
     return {
         'uuid': uuid.uuid4(),
         'start': datetime.datetime.utcnow().isoformat(),
         'timestamp': int(time.time()),
     }
 
 
 @pytest.fixture(autouse=True)
 def collect_appenlight_stats(request, testrun):
     """
     This fixture reports memory consumtion of single tests.
 
     It gathers data based on `psutil` and sends them to Appenlight. The option
     ``--ae`` has te be used to enable this fixture and the API key for your
     application has to be provided in ``--ae-key``.
     """
     try:
         # cygwin cannot have yet psutil support.
         import psutil
     except ImportError:
         return
 
     if not request.config.getoption('--appenlight'):
         return
     else:
         # Only request the baseapp fixture if appenlight tracking is
         # enabled. This will speed up a test run of unit tests by 2 to 3
         # seconds if appenlight is not enabled.
         baseapp = request.getfuncargvalue("baseapp")
     url = '{}/api/logs'.format(request.config.getoption('--appenlight-url'))
     client = AppenlightClient(
         url=url,
         api_key=request.config.getoption('--appenlight-api-key'),
         namespace=request.node.nodeid,
         request=str(testrun['uuid']),
         testrun=testrun)
 
     client.collect({
         'message': "Starting",
     })
 
     server_and_port = baseapp.config.get_settings()['vcs.server']
     protocol = baseapp.config.get_settings()['vcs.server.protocol']
     server = create_vcsserver_proxy(server_and_port, protocol)
     with server:
         vcs_pid = server.get_pid()
         server.run_gc()
     vcs_process = psutil.Process(vcs_pid)
     mem = vcs_process.memory_info()
     client.tag_before('vcsserver.rss', mem.rss)
     client.tag_before('vcsserver.vms', mem.vms)
 
     test_process = psutil.Process()
     mem = test_process.memory_info()
     client.tag_before('test.rss', mem.rss)
     client.tag_before('test.vms', mem.vms)
 
     client.tag_before('time', time.time())
 
     @request.addfinalizer
     def send_stats():
         client.tag_after('time', time.time())
         with server:
             gc_stats = server.run_gc()
         for tag, value in gc_stats.items():
             client.tag_after(tag, value)
         mem = vcs_process.memory_info()
         client.tag_after('vcsserver.rss', mem.rss)
         client.tag_after('vcsserver.vms', mem.vms)
 
         mem = test_process.memory_info()
         client.tag_after('test.rss', mem.rss)
         client.tag_after('test.vms', mem.vms)
 
         client.collect({
             'message': "Finished",
         })
         client.send_stats()
 
     return client
 
 
 class AppenlightClient():
 
     url_template = '{url}?protocol_version=0.5'
 
     def __init__(
             self, url, api_key, add_server=True, add_timestamp=True,
             namespace=None, request=None, testrun=None):
         self.url = self.url_template.format(url=url)
         self.api_key = api_key
         self.add_server = add_server
         self.add_timestamp = add_timestamp
         self.namespace = namespace
         self.request = request
         self.server = socket.getfqdn(socket.gethostname())
         self.tags_before = {}
         self.tags_after = {}
         self.stats = []
         self.testrun = testrun or {}
 
     def tag_before(self, tag, value):
         self.tags_before[tag] = value
 
     def tag_after(self, tag, value):
         self.tags_after[tag] = value
 
     def collect(self, data):
         if self.add_server:
             data.setdefault('server', self.server)
         if self.add_timestamp:
             data.setdefault('date', datetime.datetime.utcnow().isoformat())
         if self.namespace:
             data.setdefault('namespace', self.namespace)
         if self.request:
             data.setdefault('request', self.request)
         self.stats.append(data)
 
     def send_stats(self):
         tags = [
             ('testrun', self.request),
             ('testrun.start', self.testrun['start']),
             ('testrun.timestamp', self.testrun['timestamp']),
             ('test', self.namespace),
         ]
         for key, value in self.tags_before.items():
             tags.append((key + '.before', value))
             try:
                 delta = self.tags_after[key] - value
                 tags.append((key + '.delta', delta))
             except Exception:
                 pass
         for key, value in self.tags_after.items():
             tags.append((key + '.after', value))
         self.collect({
             'message': "Collected tags",
             'tags': tags,
         })
 
         response = requests.post(
             self.url,
             headers={
                 'X-appenlight-api-key': self.api_key},
             json=self.stats,
         )
 
         if not response.status_code == 200:
             pprint.pprint(self.stats)
             print(response.headers)
             print(response.text)
             raise Exception('Sending to appenlight failed')
 
 
 @pytest.fixture
 def gist_util(request, db_connection):
     """
     Provides a wired instance of `GistUtility` with integrated cleanup.
     """
     utility = GistUtility()
     request.addfinalizer(utility.cleanup)
     return utility
 
 
 class GistUtility(object):
     def __init__(self):
         self.fixture = Fixture()
         self.gist_ids = []
 
     def create_gist(self, **kwargs):
         gist = self.fixture.create_gist(**kwargs)
         self.gist_ids.append(gist.gist_id)
         return gist
 
     def cleanup(self):
         for id_ in self.gist_ids:
             self.fixture.destroy_gists(str(id_))
 
 
 @pytest.fixture
 def enabled_backends(request):
     backends = request.config.option.backends
     return backends[:]
 
 
 @pytest.fixture
 def settings_util(request, db_connection):
     """
     Provides a wired instance of `SettingsUtility` with integrated cleanup.
     """
     utility = SettingsUtility()
     request.addfinalizer(utility.cleanup)
     return utility
 
 
 class SettingsUtility(object):
     def __init__(self):
         self.rhodecode_ui_ids = []
         self.rhodecode_setting_ids = []
         self.repo_rhodecode_ui_ids = []
         self.repo_rhodecode_setting_ids = []
 
     def create_repo_rhodecode_ui(
             self, repo, section, value, key=None, active=True, cleanup=True):
         key = key or hashlib.sha1(
             '{}{}{}'.format(section, value, repo.repo_id)).hexdigest()
 
         setting = RepoRhodeCodeUi()
         setting.repository_id = repo.repo_id
         setting.ui_section = section
         setting.ui_value = value
         setting.ui_key = key
         setting.ui_active = active
         Session().add(setting)
         Session().commit()
 
         if cleanup:
             self.repo_rhodecode_ui_ids.append(setting.ui_id)
         return setting
 
     def create_rhodecode_ui(
             self, section, value, key=None, active=True, cleanup=True):
         key = key or hashlib.sha1('{}{}'.format(section, value)).hexdigest()
 
         setting = RhodeCodeUi()
         setting.ui_section = section
         setting.ui_value = value
         setting.ui_key = key
         setting.ui_active = active
         Session().add(setting)
         Session().commit()
 
         if cleanup:
             self.rhodecode_ui_ids.append(setting.ui_id)
         return setting
 
     def create_repo_rhodecode_setting(
             self, repo, name, value, type_, cleanup=True):
         setting = RepoRhodeCodeSetting(
             repo.repo_id, key=name, val=value, type=type_)
         Session().add(setting)
         Session().commit()
 
         if cleanup:
             self.repo_rhodecode_setting_ids.append(setting.app_settings_id)
         return setting
 
     def create_rhodecode_setting(self, name, value, type_, cleanup=True):
         setting = RhodeCodeSetting(key=name, val=value, type=type_)
         Session().add(setting)
         Session().commit()
 
         if cleanup:
             self.rhodecode_setting_ids.append(setting.app_settings_id)
 
         return setting
 
     def cleanup(self):
         for id_ in self.rhodecode_ui_ids:
             setting = RhodeCodeUi.get(id_)
             Session().delete(setting)
 
         for id_ in self.rhodecode_setting_ids:
             setting = RhodeCodeSetting.get(id_)
             Session().delete(setting)
 
         for id_ in self.repo_rhodecode_ui_ids:
             setting = RepoRhodeCodeUi.get(id_)
             Session().delete(setting)
 
         for id_ in self.repo_rhodecode_setting_ids:
             setting = RepoRhodeCodeSetting.get(id_)
             Session().delete(setting)
 
         Session().commit()
 
 
 @pytest.fixture
 def no_notifications(request):
     notification_patcher = mock.patch(
         'rhodecode.model.notification.NotificationModel.create')
     notification_patcher.start()
     request.addfinalizer(notification_patcher.stop)
 
 
 @pytest.fixture(scope='session')
 def repeat(request):
     """
     The number of repetitions is based on this fixture.
 
     Slower calls may divide it by 10 or 100. It is chosen in a way so that the
     tests are not too slow in our default test suite.
     """
     return request.config.getoption('--repeat')
 
 
 @pytest.fixture
 def rhodecode_fixtures():
     return Fixture()
 
 
 @pytest.fixture
 def context_stub():
     """
     Stub context object.
     """
     context = pyramid.testing.DummyResource()
     return context
 
 
 @pytest.fixture
 def request_stub():
     """
     Stub request object.
     """
     from rhodecode.lib.base import bootstrap_request
     request = bootstrap_request(scheme='https')
     return request
 
 
 @pytest.fixture
 def config_stub(request, request_stub):
     """
     Set up pyramid.testing and return the Configurator.
     """
     from rhodecode.lib.base import bootstrap_config
     config = bootstrap_config(request=request_stub)
 
     @request.addfinalizer
     def cleanup():
         pyramid.testing.tearDown()
 
     return config
 
 
 @pytest.fixture
 def StubIntegrationType():
     class _StubIntegrationType(IntegrationTypeBase):
         """ Test integration type class """
 
         key = 'test'
         display_name = 'Test integration type'
         description = 'A test integration type for testing'
 
         @classmethod
         def icon(cls):
             return 'test_icon_html_image'
 
         def __init__(self, settings):
             super(_StubIntegrationType, self).__init__(settings)
             self.sent_events = []  # for testing
 
         def send_event(self, event):
             self.sent_events.append(event)
 
         def settings_schema(self):
             class SettingsSchema(colander.Schema):
                 test_string_field = colander.SchemaNode(
                     colander.String(),
                     missing=colander.required,
                     title='test string field',
                 )
                 test_int_field = colander.SchemaNode(
                     colander.Int(),
                     title='some integer setting',
                 )
             return SettingsSchema()
 
 
     integration_type_registry.register_integration_type(_StubIntegrationType)
     return _StubIntegrationType
 
 @pytest.fixture
 def stub_integration_settings():
     return {
         'test_string_field': 'some data',
         'test_int_field': 100,
     }
 
 
 @pytest.fixture
 def repo_integration_stub(request, repo_stub, StubIntegrationType,
         stub_integration_settings):
     integration = IntegrationModel().create(
         StubIntegrationType, settings=stub_integration_settings, enabled=True,
         name='test repo integration',
         repo=repo_stub, repo_group=None, child_repos_only=None)
 
     @request.addfinalizer
     def cleanup():
         IntegrationModel().delete(integration)
 
     return integration
 
 
 @pytest.fixture
 def repogroup_integration_stub(request, test_repo_group, StubIntegrationType,
     stub_integration_settings):
     integration = IntegrationModel().create(
         StubIntegrationType, settings=stub_integration_settings, enabled=True,
         name='test repogroup integration',
         repo=None, repo_group=test_repo_group, child_repos_only=True)
 
     @request.addfinalizer
     def cleanup():
         IntegrationModel().delete(integration)
 
     return integration
 
 
 @pytest.fixture
 def repogroup_recursive_integration_stub(request, test_repo_group,
     StubIntegrationType, stub_integration_settings):
     integration = IntegrationModel().create(
         StubIntegrationType, settings=stub_integration_settings, enabled=True,
         name='test recursive repogroup integration',
         repo=None, repo_group=test_repo_group, child_repos_only=False)
 
     @request.addfinalizer
     def cleanup():
         IntegrationModel().delete(integration)
 
     return integration
 
 
 @pytest.fixture
 def global_integration_stub(request, StubIntegrationType,
     stub_integration_settings):
     integration = IntegrationModel().create(
         StubIntegrationType, settings=stub_integration_settings, enabled=True,
         name='test global integration',
         repo=None, repo_group=None, child_repos_only=None)
 
     @request.addfinalizer
     def cleanup():
         IntegrationModel().delete(integration)
 
     return integration
 
 
 @pytest.fixture
 def root_repos_integration_stub(request, StubIntegrationType,
     stub_integration_settings):
     integration = IntegrationModel().create(
         StubIntegrationType, settings=stub_integration_settings, enabled=True,
         name='test global integration',
         repo=None, repo_group=None, child_repos_only=True)
 
     @request.addfinalizer
     def cleanup():
         IntegrationModel().delete(integration)
 
     return integration
 
 
 @pytest.fixture
 def local_dt_to_utc():
     def _factory(dt):
         return dt.replace(tzinfo=dateutil.tz.tzlocal()).astimezone(
             dateutil.tz.tzutc()).replace(tzinfo=None)
     return _factory
 
 
 @pytest.fixture
 def disable_anonymous_user(request, baseapp):
     set_anonymous_access(False)
 
     @request.addfinalizer
     def cleanup():
         set_anonymous_access(True)
 
 
 @pytest.fixture(scope='module')
 def rc_fixture(request):
     return Fixture()
 
 
 @pytest.fixture
 def repo_groups(request):
     fixture = Fixture()
 
     session = Session()
     zombie_group = fixture.create_repo_group('zombie')
     parent_group = fixture.create_repo_group('parent')
     child_group = fixture.create_repo_group('parent/child')
     groups_in_db = session.query(RepoGroup).all()
     assert len(groups_in_db) == 3
     assert child_group.group_parent_id == parent_group.group_id
 
     @request.addfinalizer
     def cleanup():
         fixture.destroy_repo_group(zombie_group)
         fixture.destroy_repo_group(child_group)
         fixture.destroy_repo_group(parent_group)
 
     return zombie_group, parent_group, child_group
+
+
[e-mail](scope="session")
+def tmp_path_factory(request):
+    """Return a :class:`_pytest.tmpdir.TempPathFactory` instance for the test session.
+    """
+
+    class TempPathFactory:
+
+        def mktemp(self, basename):
+            import tempfile
+            return tempfile.mktemp(basename)
+
+    return TempPathFactory()

rhodecode/tests/vcs/test_git.py

@@ -1,1293 +1,1275 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2019 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)
+class TestGitRepository(object):
 
     @pytest.fixture(autouse=True)
     def prepare(self, request, baseapp):
         self.repo = GitRepository(TEST_GIT_REPO, bare=True)
 
-    def get_clone_repo(self):
+    def get_clone_repo(self, tmp_path_factory):
         """
         Return a non bare clone of the base repo.
         """
-        clone_path = next(REPO_PATH_GENERATOR)
+        clone_path = tmp_path_factory.mktemp('clone-url')
         repo_clone = GitRepository(
             clone_path, create=True, src_url=self.repo.path, bare=False)
 
         return repo_clone
 
-    def get_empty_repo(self, bare=False):
+    def get_empty_repo(self, tmp_path_factory, bare=False):
         """
         Return a non bare empty repo.
         """
-        return GitRepository(next(REPO_PATH_GENERATOR), create=True, bare=bare)
+        clone_path = tmp_path_factory.mktemp('empty-repo')
+        return GitRepository(clone_path, create=True, bare=bare)
 
     def test_wrong_repo_path(self):
         wrong_repo_path = '/tmp/errorrepo_git'
         with pytest.raises(RepositoryError):
             GitRepository(wrong_repo_path)
 
-    def test_repo_clone(self):
-        self.__check_for_existing_repo()
+    def test_repo_clone(self, tmp_path_factory):
         repo = GitRepository(TEST_GIT_REPO)
+        clone_path = tmp_path_factory.mktemp('_') + '_' + TEST_GIT_REPO_CLONE
         repo_clone = GitRepository(
-            TEST_GIT_REPO_CLONE,
+            clone_path,
             src_url=TEST_GIT_REPO, create=True, do_workspace_checkout=True)
+
         assert len(repo.commit_ids) == len(repo_clone.commit_ids)
         # Checking hashes of commits should be enough
         for commit in repo.get_commits():
             raw_id = commit.raw_id
             assert raw_id == repo_clone.get_commit(raw_id).raw_id
 
     def test_repo_clone_without_create(self):
         with pytest.raises(RepositoryError):
             GitRepository(
                 TEST_GIT_REPO_CLONE + '_wo_create', src_url=TEST_GIT_REPO)
 
-    def test_repo_clone_with_update(self):
+    def test_repo_clone_with_update(self, tmp_path_factory):
         repo = GitRepository(TEST_GIT_REPO)
-        clone_path = TEST_GIT_REPO_CLONE + '_with_update'
+        clone_path = tmp_path_factory.mktemp('_') + '_' + TEST_GIT_REPO_CLONE + '_update'
+
         repo_clone = GitRepository(
             clone_path,
             create=True, src_url=TEST_GIT_REPO, do_workspace_checkout=True)
         assert len(repo.commit_ids) == len(repo_clone.commit_ids)
 
         # check if current workdir was updated
         fpath = os.path.join(clone_path, 'MANIFEST.in')
         assert os.path.isfile(fpath)
 
-    def test_repo_clone_without_update(self):
+    def test_repo_clone_without_update(self, tmp_path_factory):
         repo = GitRepository(TEST_GIT_REPO)
-        clone_path = TEST_GIT_REPO_CLONE + '_without_update'
+        clone_path = tmp_path_factory.mktemp('_') + '_' + TEST_GIT_REPO_CLONE + '_without_update'
         repo_clone = GitRepository(
             clone_path,
             create=True, src_url=TEST_GIT_REPO, do_workspace_checkout=False)
         assert len(repo.commit_ids) == len(repo_clone.commit_ids)
         # check if current workdir was *NOT* updated
         fpath = os.path.join(clone_path, 'MANIFEST.in')
         # Make sure it's not bare repo
         assert not repo_clone.bare
         assert not os.path.isfile(fpath)
 
-    def test_repo_clone_into_bare_repo(self):
+    def test_repo_clone_into_bare_repo(self, tmp_path_factory):
         repo = GitRepository(TEST_GIT_REPO)
-        clone_path = TEST_GIT_REPO_CLONE + '_bare.git'
+        clone_path = tmp_path_factory.mktemp('_') + '_' + 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(bare=True)
         target_repo.fetch(source_repo.path)
         # Note: Get a fresh instance, avoids caching trouble
         target_repo = vcsbackend_git.backend(target_repo.path)
         assert len(source_repo.commit_ids) == len(target_repo.commit_ids)
 
     def test_commit_ids(self):
         # there are 112 commits (by now)
         # so we can assume they would be available from now on
         subset = {'c1214f7e79e02fc37156ff215cd71275450cffc3',
                   '38b5fe81f109cb111f549bfe9bb6b267e10bc557',
                   'fa6600f6848800641328adbf7811fd2372c02ab2',
                   '102607b09cdd60e2793929c4f90478be29f85a17',
                   '49d3fd156b6f7db46313fac355dca1a0b94a0017',
                   '2d1028c054665b962fa3d307adfc923ddd528038',
                   'd7e0d30fbcae12c90680eb095a4f5f02505ce501',
                   'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
                   'dd80b0f6cf5052f17cc738c2951c4f2070200d7f',
                   '8430a588b43b5d6da365400117c89400326e7992',
                   'd955cd312c17b02143c04fa1099a352b04368118',
                   'f67b87e5c629c2ee0ba58f85197e423ff28d735b',
                   'add63e382e4aabc9e1afdc4bdc24506c269b7618',
                   'f298fe1189f1b69779a4423f40b48edf92a703fc',
                   'bd9b619eb41994cac43d67cf4ccc8399c1125808',
                   '6e125e7c890379446e98980d8ed60fba87d0f6d1',
                   'd4a54db9f745dfeba6933bf5b1e79e15d0af20bd',
                   '0b05e4ed56c802098dfc813cbe779b2f49e92500',
                   '191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
                   '45223f8f114c64bf4d6f853e3c35a369a6305520',
                   'ca1eb7957a54bce53b12d1a51b13452f95bc7c7e',
                   'f5ea29fc42ef67a2a5a7aecff10e1566699acd68',
                   '27d48942240f5b91dfda77accd2caac94708cc7d',
                   '622f0eb0bafd619d2560c26f80f09e3b0b0d78af',
                   'e686b958768ee96af8029fe19c6050b1a8dd3b2b'}
         assert subset.issubset(set(self.repo.commit_ids))
 
     def test_slicing(self):
         # 4 1 5 10 95
         for sfrom, sto, size in [(0, 4, 4), (1, 2, 1), (10, 15, 5),
                                  (10, 20, 10), (5, 100, 95)]:
             commit_ids = list(self.repo[sfrom:sto])
             assert len(commit_ids) == size
             assert commit_ids[0] == self.repo.get_commit(commit_idx=sfrom)
             assert commit_ids[-1] == self.repo.get_commit(commit_idx=sto - 1)
 
     def test_branches(self):
         # TODO: Need more tests here
         # Removed (those are 'remotes' branches for cloned repo)
         # assert 'master' in self.repo.branches
         # assert 'gittree' in self.repo.branches
         # assert 'web-branch' in self.repo.branches
         for __, commit_id in self.repo.branches.items():
             assert isinstance(self.repo.get_commit(commit_id), GitCommit)
 
     def test_tags(self):
         # TODO: Need more tests here
         assert 'v0.1.1' in self.repo.tags
         assert 'v0.1.2' in self.repo.tags
         for __, commit_id in self.repo.tags.items():
             assert isinstance(self.repo.get_commit(commit_id), GitCommit)
 
     def _test_single_commit_cache(self, commit_id):
         commit = self.repo.get_commit(commit_id)
         assert commit_id in self.repo.commits
         assert commit is self.repo.commits[commit_id]
 
     def test_initial_commit(self):
         commit_id = self.repo.commit_ids[0]
         init_commit = self.repo.get_commit(commit_id)
         init_author = init_commit.author
 
         assert init_commit.message == 'initial import\n'
         assert init_author == 'Marcin Kuzminski <[e-mail]>'
         assert init_author == init_commit.committer
         for path in ('vcs/__init__.py',
                      'vcs/backends/BaseRepository.py',
                      'vcs/backends/__init__.py'):
             assert isinstance(init_commit.get_node(path), FileNode)
         for path in ('', 'vcs', 'vcs/backends'):
             assert isinstance(init_commit.get_node(path), DirNode)
 
         with pytest.raises(NodeDoesNotExistError):
             init_commit.get_node(path='foobar')
 
         node = init_commit.get_node('vcs/')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.DIR
 
         node = init_commit.get_node('vcs')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.DIR
 
         node = init_commit.get_node('vcs/__init__.py')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.FILE
 
     def test_not_existing_commit(self):
         with pytest.raises(RepositoryError):
             self.repo.get_commit('f' * 40)
 
     def test_commit10(self):
 
         commit10 = self.repo.get_commit(self.repo.commit_ids[9])
         README = """===
 VCS
 ===
 
 Various Version Control System management abstraction layer for Python.
 
 Introduction
 ------------
 
 TODO: To be written...
 
 """
         node = commit10.get_node('README.rst')
         assert node.kind == NodeKind.FILE
         assert node.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()
+    def test_checkout_with_create(self, tmp_path_factory):
+        repo_clone = self.get_clone_repo(tmp_path_factory)
 
         new_branch = 'new_branch'
         assert repo_clone._current_branch() == 'master'
         assert set(repo_clone.branches) == {'master'}
         repo_clone._checkout(new_branch, create=True)
 
         # Branches is a lazy property so we need to recrete the Repo object.
         repo_clone = GitRepository(repo_clone.path)
         assert set(repo_clone.branches) == {'master', new_branch}
         assert repo_clone._current_branch() == new_branch
 
-    def test_checkout(self):
-        repo_clone = self.get_clone_repo()
+    def test_checkout(self, tmp_path_factory):
+        repo_clone = self.get_clone_repo(tmp_path_factory)
 
         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()
+    def test_checkout_same_branch(self, tmp_path_factory):
+        repo_clone = self.get_clone_repo(tmp_path_factory)
 
         repo_clone._checkout('master')
         assert repo_clone._current_branch() == 'master'
 
-    def test_checkout_branch_already_exists(self):
-        repo_clone = self.get_clone_repo()
+    def test_checkout_branch_already_exists(self, tmp_path_factory):
+        repo_clone = self.get_clone_repo(tmp_path_factory)
 
         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()
+    def test_current_branch_empty_repo(self, tmp_path_factory):
+        repo = self.get_empty_repo(tmp_path_factory)
         assert repo._current_branch() is None
 
-    def test_local_clone(self):
-        clone_path = next(REPO_PATH_GENERATOR)
+    def test_local_clone(self, tmp_path_factory):
+        clone_path = tmp_path_factory.mktemp('test-local-clone')
         self.repo._local_clone(clone_path, 'master')
         repo_clone = GitRepository(clone_path)
 
         assert self.repo.commit_ids == repo_clone.commit_ids
 
-    def test_local_clone_with_specific_branch(self):
-        source_repo = self.get_clone_repo()
+    def test_local_clone_with_specific_branch(self, tmp_path_factory):
+        source_repo = self.get_clone_repo(tmp_path_factory)
 
         # 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)
+        clone_path = tmp_path_factory.mktemp('git-clone-path-1')
         source_repo._local_clone(clone_path, 'new_branch')
         repo_clone = GitRepository(clone_path)
 
         assert source_repo.commit_ids[:-3 + 1] == repo_clone.commit_ids
 
-        clone_path = next(REPO_PATH_GENERATOR)
+        clone_path = tmp_path_factory.mktemp('git-clone-path-2')
         source_repo._local_clone(clone_path, 'master')
         repo_clone = GitRepository(clone_path)
 
         assert source_repo.commit_ids == repo_clone.commit_ids
 
     def test_local_clone_fails_if_target_exists(self):
         with pytest.raises(RepositoryError):
             self.repo._local_clone(self.repo.path, 'master')
 
-    def test_local_fetch(self):
-        target_repo = self.get_empty_repo()
-        source_repo = self.get_clone_repo()
+    def test_local_fetch(self, tmp_path_factory):
+        target_repo = self.get_empty_repo(tmp_path_factory)
+        source_repo = self.get_clone_repo(tmp_path_factory)
 
         # 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()
+    def test_local_fetch_from_bare_repo(self, tmp_path_factory):
+        target_repo = self.get_empty_repo(tmp_path_factory)
         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()
+    def test_local_fetch_branch_does_not_exist(self, tmp_path_factory):
+        target_repo = self.get_empty_repo(tmp_path_factory)
 
         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()
+    def test_local_pull(self, tmp_path_factory):
+        target_repo = self.get_empty_repo(tmp_path_factory)
+        source_repo = self.get_clone_repo(tmp_path_factory)
 
         # 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()
+    def test_local_merge(self, tmp_path_factory):
+        target_repo = self.get_empty_repo(tmp_path_factory)
+        source_repo = self.get_clone_repo(tmp_path_factory)
 
         # 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()
+    def test_local_merge_into_empty_repo(self, tmp_path_factory):
+        target_repo = self.get_empty_repo(tmp_path_factory)
 
         # 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()
+    def test_local_push_non_bare(self, tmp_path_factory):
+        target_repo = self.get_empty_repo(tmp_path_factory)
 
         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)
+    def test_local_push_bare(self, tmp_path_factory):
+        target_repo = self.get_empty_repo(tmp_path_factory, 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()
+    def test_local_push_non_bare_target_branch_is_checked_out(self, tmp_path_factory):
+        target_repo = self.get_clone_repo(tmp_path_factory)
 
         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)
+    def test_hooks_can_be_enabled_via_env_variable_for_local_push(self, tmp_path_factory):
+        target_repo = self.get_empty_repo(tmp_path_factory, bare=True)
 
         with mock.patch.object(self.repo, 'run_git_command') as run_mock:
             self.repo._local_push(
                 'master', target_repo.path, 'master', enable_hooks=True)
         env = run_mock.call_args[1]['extra_env']
         assert 'RC_SKIP_HOOKS' not in env
 
     def _add_failing_hook(self, repo_path, hook_name, bare=False):
         path_components = (
             ['hooks', hook_name] if bare else ['.git', 'hooks', hook_name])
         hook_path = os.path.join(repo_path, *path_components)
         with open(hook_path, 'w') as f:
             script_lines = [
                 '#!%s' % sys.executable,
                 'import os',
                 'import sys',
                 'if os.environ.get("RC_SKIP_HOOKS"):',
                 '    sys.exit(0)',
                 'sys.exit(1)',
             ]
             f.write('\n'.join(script_lines))
         os.chmod(hook_path, 0o755)
 
-    def test_local_push_does_not_execute_hook(self):
-        target_repo = self.get_empty_repo()
+    def test_local_push_does_not_execute_hook(self, tmp_path_factory):
+        target_repo = self.get_empty_repo(tmp_path_factory)
 
         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)
+    def test_local_push_executes_hook(self, tmp_path_factory):
+        target_repo = self.get_empty_repo(tmp_path_factory, bare=True)
         self._add_failing_hook(target_repo.path, 'pre-receive', bare=True)
         with pytest.raises(RepositoryError):
             self.repo._local_push(
                 'master', target_repo.path, 'master', enable_hooks=True)
 
     def test_maybe_prepare_merge_workspace(self):
         workspace = self.repo._maybe_prepare_merge_workspace(
             2, 'pr2', Reference('branch', 'master', 'unused'),
             Reference('branch', 'master', 'unused'))
 
         assert os.path.isdir(workspace)
         workspace_repo = GitRepository(workspace)
         assert workspace_repo.branches == self.repo.branches
 
         # Calling it a second time should also succeed
         workspace = self.repo._maybe_prepare_merge_workspace(
             2, 'pr2', Reference('branch', 'master', 'unused'),
             Reference('branch', 'master', 'unused'))
         assert os.path.isdir(workspace)
 
     def test_maybe_prepare_merge_workspace_different_refs(self):
         workspace = self.repo._maybe_prepare_merge_workspace(
             2, 'pr2', Reference('branch', 'master', 'unused'),
             Reference('branch', 'develop', 'unused'))
 
         assert os.path.isdir(workspace)
         workspace_repo = GitRepository(workspace)
         assert workspace_repo.branches == self.repo.branches
 
         # Calling it a second time should also succeed
         workspace = self.repo._maybe_prepare_merge_workspace(
             2, 'pr2', Reference('branch', 'master', 'unused'),
             Reference('branch', 'develop', 'unused'))
         assert os.path.isdir(workspace)
 
     def test_cleanup_merge_workspace(self):
         workspace = self.repo._maybe_prepare_merge_workspace(
             2, 'pr3', Reference('branch', 'master', 'unused'),
             Reference('branch', 'master', 'unused'))
         self.repo.cleanup_merge_workspace(2, 'pr3')
 
         assert not os.path.exists(workspace)
 
     def test_cleanup_merge_workspace_invalid_workspace_id(self):
         # No assert: because in case of an inexistent workspace this function
         # should still succeed.
         self.repo.cleanup_merge_workspace(1, 'pr4')
 
     def test_set_refs(self):
         test_ref = 'refs/test-refs/abcde'
         test_commit_id = 'ecb86e1f424f2608262b130db174a7dfd25a6623'
 
         self.repo.set_refs(test_ref, test_commit_id)
         stdout, _ = self.repo.run_git_command(['show-ref'])
         assert test_ref in stdout
         assert test_commit_id in stdout
 
     def test_remove_ref(self):
         test_ref = 'refs/test-refs/abcde'
         test_commit_id = 'ecb86e1f424f2608262b130db174a7dfd25a6623'
         self.repo.set_refs(test_ref, test_commit_id)
         stdout, _ = self.repo.run_git_command(['show-ref'])
         assert test_ref in stdout
         assert test_commit_id in stdout
 
         self.repo.remove_ref(test_ref)
         stdout, _ = self.repo.run_git_command(['show-ref'])
         assert test_ref not in stdout
         assert test_commit_id not in stdout
 
 
 class TestGitCommit(object):
 
     @pytest.fixture(autouse=True)
     def prepare(self):
         self.repo = GitRepository(TEST_GIT_REPO)
 
     def test_default_commit(self):
         tip = self.repo.get_commit()
         assert tip == self.repo.get_commit(None)
         assert tip == self.repo.get_commit('tip')
 
     def test_root_node(self):
         tip = self.repo.get_commit()
         assert tip.root is tip.get_node('')
 
     def test_lazy_fetch(self):
         """
         Test if commit's nodes expands and are cached as we walk through
         the commit. This test is somewhat hard to write as order of tests
         is a key here. Written by running command after command in a shell.
         """
         commit_id = '2a13f185e4525f9d4b59882791a2d397b90d5ddc'
         assert commit_id in self.repo.commit_ids
         commit = self.repo.get_commit(commit_id)
         assert len(commit.nodes) == 0
         root = commit.root
         assert len(commit.nodes) == 1
         assert len(root.nodes) == 8
         # accessing root.nodes updates commit.nodes
         assert len(commit.nodes) == 9
 
         docs = root.get_node('docs')
         # we haven't yet accessed anything new as docs dir was already cached
         assert len(commit.nodes) == 9
         assert len(docs.nodes) == 8
         # accessing docs.nodes updates commit.nodes
         assert len(commit.nodes) == 17
 
         assert docs is commit.get_node('docs')
         assert docs is root.nodes[0]
         assert docs is root.dirs[0]
         assert docs is commit.get_node('docs')
 
     def test_nodes_with_commit(self):
         commit_id = '2a13f185e4525f9d4b59882791a2d397b90d5ddc'
         commit = self.repo.get_commit(commit_id)
         root = commit.root
         docs = root.get_node('docs')
         assert docs is commit.get_node('docs')
         api = docs.get_node('api')
         assert api is commit.get_node('docs/api')
         index = api.get_node('index.rst')
         assert index is commit.get_node('docs/api/index.rst')
         assert index is commit.get_node('docs')\
             .get_node('api')\
             .get_node('index.rst')
 
     def test_branch_and_tags(self):
         """
         rev0 = self.repo.commit_ids[0]
         commit0 = self.repo.get_commit(rev0)
         assert commit0.branch == 'master'
         assert commit0.tags == []
 
         rev10 = self.repo.commit_ids[10]
         commit10 = self.repo.get_commit(rev10)
         assert commit10.branch == 'master'
         assert commit10.tags == []
 
         rev44 = self.repo.commit_ids[44]
         commit44 = self.repo.get_commit(rev44)
         assert commit44.branch == 'web-branch'
 
         tip = self.repo.get_commit('tip')
         assert 'tip' in tip.tags
         """
         # Those tests would fail - branches are now going
         # to be changed at main API in order to support git backend
         pass
 
     def test_file_size(self):
         to_check = (
             ('c1214f7e79e02fc37156ff215cd71275450cffc3',
                 'vcs/backends/BaseRepository.py', 502),
             ('d7e0d30fbcae12c90680eb095a4f5f02505ce501',
                 'vcs/backends/hg.py', 854),
             ('6e125e7c890379446e98980d8ed60fba87d0f6d1',
                 'setup.py', 1068),
 
             ('d955cd312c17b02143c04fa1099a352b04368118',
                 'vcs/backends/base.py', 2921),
             ('ca1eb7957a54bce53b12d1a51b13452f95bc7c7e',
                 'vcs/backends/base.py', 3936),
             ('f50f42baeed5af6518ef4b0cb2f1423f3851a941',
                 'vcs/backends/base.py', 6189),
         )
         for commit_id, path, size in to_check:
             node = self.repo.get_commit(commit_id).get_node(path)
             assert node.is_file()
             assert node.size == size
 
     def test_file_history_from_commits(self):
         node = self.repo[10].get_node('setup.py')
         commit_ids = [commit.raw_id for commit in node.history]
         assert ['ff7ca51e58c505fec0dd2491de52c622bb7a806b'] == commit_ids
 
         node = self.repo[20].get_node('setup.py')
         node_ids = [commit.raw_id for commit in node.history]
         assert ['191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b'] == node_ids
 
         # special case we check history from commit that has this particular
         # file changed this means we check if it's included as well
         node = self.repo.get_commit('191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e') \
             .get_node('setup.py')
         node_ids = [commit.raw_id for commit in node.history]
         assert ['191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b'] == node_ids
 
     def test_file_history(self):
         # we can only check if those commits are present in the history
         # as we cannot update this test every time file is changed
         files = {
             'setup.py': [
                 '54386793436c938cff89326944d4c2702340037d',
                 '51d254f0ecf5df2ce50c0b115741f4cf13985dab',
                 '998ed409c795fec2012b1c0ca054d99888b22090',
                 '5e0eb4c47f56564395f76333f319d26c79e2fb09',
                 '0115510b70c7229dbc5dc49036b32e7d91d23acd',
                 '7cb3fd1b6d8c20ba89e2264f1c8baebc8a52d36e',
                 '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                 '191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             ],
             'vcs/nodes.py': [
                 '33fa3223355104431402a888fa77a4e9956feb3e',
                 'fa014c12c26d10ba682fadb78f2a11c24c8118e1',
                 'e686b958768ee96af8029fe19c6050b1a8dd3b2b',
                 'ab5721ca0a081f26bf43d9051e615af2cc99952f',
                 'c877b68d18e792a66b7f4c529ea02c8f80801542',
                 '4313566d2e417cb382948f8d9d7c765330356054',
                 '6c2303a793671e807d1cfc70134c9ca0767d98c2',
                 '54386793436c938cff89326944d4c2702340037d',
                 '54000345d2e78b03a99d561399e8e548de3f3203',
                 '1c6b3677b37ea064cb4b51714d8f7498f93f4b2b',
                 '2d03ca750a44440fb5ea8b751176d1f36f8e8f46',
                 '2a08b128c206db48c2f0b8f70df060e6db0ae4f8',
                 '30c26513ff1eb8e5ce0e1c6b477ee5dc50e2f34b',
                 'ac71e9503c2ca95542839af0ce7b64011b72ea7c',
                 '12669288fd13adba2a9b7dd5b870cc23ffab92d2',
                 '5a0c84f3e6fe3473e4c8427199d5a6fc71a9b382',
                 '12f2f5e2b38e6ff3fbdb5d722efed9aa72ecb0d5',
                 '5eab1222a7cd4bfcbabc218ca6d04276d4e27378',
                 'f50f42baeed5af6518ef4b0cb2f1423f3851a941',
                 'd7e390a45f6aa96f04f5e7f583ad4f867431aa25',
                 'f15c21f97864b4f071cddfbf2750ec2e23859414',
                 'e906ef056cf539a4e4e5fc8003eaf7cf14dd8ade',
                 'ea2b108b48aa8f8c9c4a941f66c1a03315ca1c3b',
                 '84dec09632a4458f79f50ddbbd155506c460b4f9',
                 '0115510b70c7229dbc5dc49036b32e7d91d23acd',
                 '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                 '3bf1c5868e570e39569d094f922d33ced2fa3b2b',
                 'b8d04012574729d2c29886e53b1a43ef16dd00a1',
                 '6970b057cffe4aab0a792aa634c89f4bebf01441',
                 'dd80b0f6cf5052f17cc738c2951c4f2070200d7f',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             ],
             'vcs/backends/git.py': [
                 '4cf116ad5a457530381135e2f4c453e68a1b0105',
                 '9a751d84d8e9408e736329767387f41b36935153',
                 'cb681fb539c3faaedbcdf5ca71ca413425c18f01',
                 '428f81bb652bcba8d631bce926e8834ff49bdcc6',
                 '180ab15aebf26f98f714d8c68715e0f05fa6e1c7',
                 '2b8e07312a2e89e92b90426ab97f349f4bce2a3a',
                 '50e08c506174d8645a4bb517dd122ac946a0f3bf',
                 '54000345d2e78b03a99d561399e8e548de3f3203',
             ],
         }
         for path, commit_ids in files.items():
             node = self.repo.get_commit(commit_ids[0]).get_node(path)
             node_ids = [commit.raw_id for commit in node.history]
             assert set(commit_ids).issubset(set(node_ids)), (
                 "We assumed that %s is subset of commit_ids for which file %s "
                 "has been changed, and history of that node returned: %s"
                 % (commit_ids, path, node_ids))
 
     def test_file_annotate(self):
         files = {
             'vcs/backends/__init__.py': {
                 'c1214f7e79e02fc37156ff215cd71275450cffc3': {
                     'lines_no': 1,
                     'commits': [
                         'c1214f7e79e02fc37156ff215cd71275450cffc3',
                     ],
                 },
                 '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647': {
                     'lines_no': 21,
                     'commits': [
                         '49d3fd156b6f7db46313fac355dca1a0b94a0017',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                     ],
                 },
                 'e29b67bd158580fc90fc5e9111240b90e6e86064': {
                     'lines_no': 32,
                     'commits': [
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '5eab1222a7cd4bfcbabc218ca6d04276d4e27378',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '54000345d2e78b03a99d561399e8e548de3f3203',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '78c3f0c23b7ee935ec276acb8b8212444c33c396',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '78c3f0c23b7ee935ec276acb8b8212444c33c396',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                     ],
                 },
             },
         }
 
         for fname, commit_dict in files.items():
             for commit_id, __ in commit_dict.items():
                 commit = self.repo.get_commit(commit_id)
 
                 l1_1 = [x[1] for x in commit.get_file_annotate(fname)]
                 l1_2 = [x[2]().raw_id for x in commit.get_file_annotate(fname)]
                 assert l1_1 == l1_2
                 l1 = l1_1
                 l2 = files[fname][commit_id]['commits']
                 assert l1 == l2, (
                     "The lists of commit_ids for %s@commit_id %s"
                     "from annotation list should match each other, "
                     "got \n%s \nvs \n%s " % (fname, commit_id, l1, l2))
 
     def test_files_state(self):
         """
         Tests state of FileNodes.
         """
         node = self.repo\
             .get_commit('e6ea6d16e2f26250124a1f4b4fe37a912f9d86a0')\
             .get_node('vcs/utils/diffs.py')
         assert node.state, NodeState.ADDED
         assert node.added
         assert not node.changed
         assert not node.not_changed
         assert not node.removed
 
         node = self.repo\
             .get_commit('33fa3223355104431402a888fa77a4e9956feb3e')\
             .get_node('.hgignore')
         assert node.state, NodeState.CHANGED
         assert not node.added
         assert node.changed
         assert not node.not_changed
         assert not node.removed
 
         node = self.repo\
             .get_commit('e29b67bd158580fc90fc5e9111240b90e6e86064')\
             .get_node('setup.py')
         assert node.state, NodeState.NOT_CHANGED
         assert not node.added
         assert not node.changed
         assert node.not_changed
         assert not node.removed
 
         # If node has REMOVED state then trying to fetch it would raise
         # CommitError exception
         commit = self.repo.get_commit(
             'fa6600f6848800641328adbf7811fd2372c02ab2')
         path = 'vcs/backends/BaseRepository.py'
         with pytest.raises(NodeDoesNotExistError):
             commit.get_node(path)
         # but it would be one of ``removed`` (commit's attribute)
         assert path in [rf.path for rf in commit.removed]
 
         commit = self.repo.get_commit(
             '54386793436c938cff89326944d4c2702340037d')
         changed = [
             'setup.py', 'tests/test_nodes.py', 'vcs/backends/hg.py',
             'vcs/nodes.py']
         assert set(changed) == set([f.path for f in commit.changed])
 
     def test_unicode_branch_refs(self):
         unicode_branches = {
             'refs/heads/unicode': '6c0ce52b229aa978889e91b38777f800e85f330b',
             u'refs/heads/uniçö∂e': 'ürl',
         }
         with mock.patch(
             ("rhodecode.lib.vcs.backends.git.repository"
                 ".GitRepository._refs"),
                 unicode_branches):
             branches = self.repo.branches
 
         assert 'unicode' in branches
         assert u'uniçö∂e' in branches
 
     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=0o120000),  # 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):
         comm1 = self.repo[0]
         comm2 = self.repo[1]
         self.repo.run_git_command = mock.Mock(return_value=['', ''])
         self.repo.get_diff(comm1, comm2)
 
         self.repo.run_git_command.assert_called_once_with(
             ['diff', '-U3', '--full-index', '--binary', '-p', '-M',
              '--abbrev=40', comm1.raw_id, comm2.raw_id])
 
     def test_get_diff_runs_git_command_with_str_hashes(self):
         comm2 = self.repo[1]
         self.repo.run_git_command = mock.Mock(return_value=['', ''])
         self.repo.get_diff(self.repo.EMPTY_COMMIT, comm2)
         self.repo.run_git_command.assert_called_once_with(
             ['show', '-U3', '--full-index', '--binary', '-p', '-M',
              '--abbrev=40', comm2.raw_id])
 
     def test_get_diff_runs_git_command_with_path_if_its_given(self):
         comm1 = self.repo[0]
         comm2 = self.repo[1]
         self.repo.run_git_command = mock.Mock(return_value=['', ''])
         self.repo.get_diff(comm1, comm2, 'foo')
         self.repo.run_git_command.assert_called_once_with(
             ['diff', '-U3', '--full-index', '--binary', '-p', '-M',
              '--abbrev=40', self.repo._lookup_commit(0),
              comm2.raw_id, '--', '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(object):
 
     def test_returns_git_version(self, baseapp):
         version = discover_git_version()
         assert version
 
     def test_returns_empty_string_without_vcsserver(self):
         mock_connection = mock.Mock()
         mock_connection.discover_git_version = mock.Mock(
             side_effect=Exception)
         with mock.patch('rhodecode.lib.vcs.connection.Git', mock_connection):
             version = discover_git_version()
         assert version == ''
 
 
 class TestGetSubmoduleUrl(object):
     def test_submodules_file_found(self):
         commit = GitCommit(repository=mock.Mock(), raw_id='abcdef12', idx=1)
         node = mock.Mock()
         with mock.patch.object(
                 commit, 'get_node', return_value=node) as get_node_mock:
             node.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_operations/test_vcs_operations_special.py

@@ -1,281 +1,282 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2019 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 pytest
 
 from rhodecode.lib.vcs.backends.git.repository import GitRepository
 from rhodecode.lib.vcs.backends.hg.repository import MercurialRepository
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.model.db import Repository
 from rhodecode.model.meta import Session
 from rhodecode.tests import GIT_REPO, HG_REPO
 
 from rhodecode.tests.vcs_operations import (
     Command, _check_proper_clone, _check_proper_git_push, _check_proper_hg_push,
     _add_files_and_push)
 
 
 @pytest.mark.usefixtures("disable_locking")
 class TestVCSOperationsSpecial(object):
 
     def test_git_sets_default_branch_if_not_master(
             self, backend_git, tmpdir, rc_web_server):
         empty_repo = backend_git.create_repo()
         clone_url = rc_web_server.repo_clone_url(empty_repo.repo_name)
 
         cmd = Command(tmpdir.strpath)
         cmd.execute('git clone', clone_url)
 
         repo = GitRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
+        repo._checkout('test', create=True)
         repo.in_memory_commit.add(FileNode('file', content=''))
         repo.in_memory_commit.commit(
             message='Commit on branch test',
             author='Automatic test',
             branch='test')
 
         repo_cmd = Command(repo.path)
         stdout, stderr = repo_cmd.execute('git push --verbose origin test')
         _check_proper_git_push(
             stdout, stderr, branch='test', should_set_default_branch=True)
 
         stdout, stderr = cmd.execute(
             'git clone', clone_url, empty_repo.repo_name + '-clone')
         _check_proper_clone(stdout, stderr, 'git')
 
         # Doing an explicit commit in order to get latest user logs on MySQL
         Session().commit()
 
     def test_git_fetches_from_remote_repository_with_annotated_tags(
             self, backend_git, rc_web_server):
         # Note: This is a test specific to the git backend. It checks the
         # integration of fetching from a remote repository which contains
         # annotated tags.
 
         # Dulwich shows this specific behavior only when
         # operating against a remote repository.
         source_repo = backend_git['annotated-tag']
         target_vcs_repo = backend_git.create_repo().scm_instance()
         target_vcs_repo.fetch(rc_web_server.repo_clone_url(source_repo.repo_name))
 
     def test_git_push_shows_pull_request_refs(self, backend_git, rc_web_server, tmpdir):
         """
         test if remote info about refs is visible
         """
         empty_repo = backend_git.create_repo()
 
         clone_url = rc_web_server.repo_clone_url(empty_repo.repo_name)
 
         cmd = Command(tmpdir.strpath)
         cmd.execute('git clone', clone_url)
 
         repo = GitRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
         repo.in_memory_commit.add(FileNode('readme.md', content='## Hello'))
         repo.in_memory_commit.commit(
             message='Commit on branch Master',
             author='Automatic test',
             branch='master')
 
         repo_cmd = Command(repo.path)
         stdout, stderr = repo_cmd.execute('git push --verbose origin master')
         _check_proper_git_push(stdout, stderr, branch='master')
 
         ref = '{}/{}/pull-request/new?branch=master'.format(
             rc_web_server.host_url(), empty_repo.repo_name)
         assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stderr
         assert 'remote: RhodeCode: push completed' in stderr
 
         # push on the same branch
         repo = GitRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
         repo.in_memory_commit.add(FileNode('setup.py', content='print\n'))
         repo.in_memory_commit.commit(
             message='Commit2 on branch Master',
             author='Automatic test2',
             branch='master')
 
         repo_cmd = Command(repo.path)
         stdout, stderr = repo_cmd.execute('git push --verbose origin master')
         _check_proper_git_push(stdout, stderr, branch='master')
 
         assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stderr
         assert 'remote: RhodeCode: push completed' in stderr
 
         # new Branch
         repo = GitRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
         repo.in_memory_commit.add(FileNode('feature1.py', content='## Hello world'))
         repo.in_memory_commit.commit(
             message='Commit on branch feature',
             author='Automatic test',
             branch='feature')
 
         repo_cmd = Command(repo.path)
         stdout, stderr = repo_cmd.execute('git push --verbose origin feature')
         _check_proper_git_push(stdout, stderr, branch='feature')
 
         ref = '{}/{}/pull-request/new?branch=feature'.format(
             rc_web_server.host_url(), empty_repo.repo_name)
         assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stderr
         assert 'remote: RhodeCode: push completed' in stderr
 
     def test_hg_push_shows_pull_request_refs(self, backend_hg, rc_web_server, tmpdir):
         empty_repo = backend_hg.create_repo()
 
         clone_url = rc_web_server.repo_clone_url(empty_repo.repo_name)
 
         cmd = Command(tmpdir.strpath)
         cmd.execute('hg clone', clone_url)
 
         repo = MercurialRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
         repo.in_memory_commit.add(FileNode(u'readme.md', content=u'## Hello'))
         repo.in_memory_commit.commit(
             message=u'Commit on branch default',
             author=u'Automatic test',
             branch='default')
 
         repo_cmd = Command(repo.path)
         repo_cmd.execute('hg checkout default')
 
         stdout, stderr = repo_cmd.execute('hg push --verbose', clone_url)
         _check_proper_hg_push(stdout, stderr, branch='default')
 
         ref = '{}/{}/pull-request/new?branch=default'.format(
             rc_web_server.host_url(), empty_repo.repo_name)
         assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
         assert 'remote: RhodeCode: push completed' in stdout
 
         # push on the same branch
         repo = MercurialRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
         repo.in_memory_commit.add(FileNode(u'setup.py', content=u'print\n'))
         repo.in_memory_commit.commit(
             message=u'Commit2 on branch default',
             author=u'Automatic test2',
             branch=u'default')
 
         repo_cmd = Command(repo.path)
         repo_cmd.execute('hg checkout default')
 
         stdout, stderr = repo_cmd.execute('hg push --verbose', clone_url)
         _check_proper_hg_push(stdout, stderr, branch='default')
 
         assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
         assert 'remote: RhodeCode: push completed' in stdout
 
         # new Branch
         repo = MercurialRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
         repo.in_memory_commit.add(FileNode(u'feature1.py', content=u'## Hello world'))
         repo.in_memory_commit.commit(
             message=u'Commit on branch feature',
             author=u'Automatic test',
             branch=u'feature')
 
         repo_cmd = Command(repo.path)
         repo_cmd.execute('hg checkout feature')
 
         stdout, stderr = repo_cmd.execute('hg push --new-branch --verbose', clone_url)
         _check_proper_hg_push(stdout, stderr, branch='feature')
 
         ref = '{}/{}/pull-request/new?branch=feature'.format(
             rc_web_server.host_url(), empty_repo.repo_name)
         assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
         assert 'remote: RhodeCode: push completed' in stdout
 
     def test_hg_push_shows_pull_request_refs_book(self, backend_hg, rc_web_server, tmpdir):
         empty_repo = backend_hg.create_repo()
 
         clone_url = rc_web_server.repo_clone_url(empty_repo.repo_name)
 
         cmd = Command(tmpdir.strpath)
         cmd.execute('hg clone', clone_url)
 
         repo = MercurialRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
         repo.in_memory_commit.add(FileNode(u'readme.md', content=u'## Hello'))
         repo.in_memory_commit.commit(
             message=u'Commit on branch default',
             author=u'Automatic test',
             branch='default')
 
         repo_cmd = Command(repo.path)
         repo_cmd.execute('hg checkout default')
 
         stdout, stderr = repo_cmd.execute('hg push --verbose', clone_url)
         _check_proper_hg_push(stdout, stderr, branch='default')
 
         ref = '{}/{}/pull-request/new?branch=default'.format(
             rc_web_server.host_url(), empty_repo.repo_name)
         assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
         assert 'remote: RhodeCode: push completed' in stdout
 
         # add bookmark
         repo = MercurialRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
         repo.in_memory_commit.add(FileNode(u'setup.py', content=u'print\n'))
         repo.in_memory_commit.commit(
             message=u'Commit2 on branch default',
             author=u'Automatic test2',
             branch=u'default')
 
         repo_cmd = Command(repo.path)
         repo_cmd.execute('hg checkout default')
         repo_cmd.execute('hg bookmark feature2')
         stdout, stderr = repo_cmd.execute('hg push -B feature2 --verbose', clone_url)
         _check_proper_hg_push(stdout, stderr, branch='default')
 
         ref = '{}/{}/pull-request/new?branch=default'.format(
             rc_web_server.host_url(), empty_repo.repo_name)
         assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
         ref = '{}/{}/pull-request/new?bookmark=feature2'.format(
             rc_web_server.host_url(), empty_repo.repo_name)
         assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
         assert 'remote: RhodeCode: push completed' in stdout
         assert 'exporting bookmark feature2' in stdout
 
     def test_push_is_forbidden_on_archived_repo_hg(self, backend_hg, rc_web_server, tmpdir):
         empty_repo = backend_hg.create_repo()
         repo_name = empty_repo.repo_name
 
         repo = Repository.get_by_repo_name(repo_name)
         repo.archived = True
         Session().commit()
 
         clone_url = rc_web_server.repo_clone_url(repo_name)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
 
         stdout, stderr = _add_files_and_push(
             'hg', tmpdir.strpath, clone_url=clone_url)
 
         assert 'abort: HTTP Error 403: Forbidden' in stderr
 
     def test_push_is_forbidden_on_archived_repo_git(self, backend_git, rc_web_server, tmpdir):
         empty_repo = backend_git.create_repo()
         repo_name = empty_repo.repo_name
 
         repo = Repository.get_by_repo_name(repo_name)
         repo.archived = True
         Session().commit()
 
         clone_url = rc_web_server.repo_clone_url(repo_name)
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
 
         stdout, stderr = _add_files_and_push(
             'git', tmpdir.strpath, clone_url=clone_url)
 
         assert "The requested URL returned error: 403" in stderr