fix(tests): fixed tests for PR celery hooks deamon

rhodecode/lib/hook_daemon/base.py

@@ -1,89 +1,94 b''
 # Copyright (C) 2010-2023 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 time
 import logging
 
-from rhodecode.lib.config_utils import get_config
+from rhodecode.lib.config_utils import get_app_config_lightweight
 
 from rhodecode.lib.svn_txn_utils import get_txn_id_from_store
 
 log = logging.getLogger(__name__)
 
 
 class BaseHooksCallbackDaemon:
     """
     Basic context manager for actions that don't require some extra
     """
     def __init__(self):
         pass
 
     def __enter__(self):
         log.debug('Running `%s` callback daemon', self.__class__.__name__)
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         log.debug('Exiting `%s` callback daemon', self.__class__.__name__)
 
 
 class HooksModuleCallbackDaemon(BaseHooksCallbackDaemon):
 
     def __init__(self, module):
         super().__init__()
         self.hooks_module = module
 
     def __repr__(self):
         return f'HooksModuleCallbackDaemon(hooks_module={self.hooks_module})'
 
 
 def prepare_callback_daemon(extras, protocol, host, txn_id=None):
 
     match protocol:
         case 'http':
             from rhodecode.lib.hook_daemon.http_hooks_deamon import HttpHooksCallbackDaemon
             port = 0
             if txn_id:
                 # read txn-id to re-use the PORT for callback daemon
                 repo_path = os.path.join(extras['repo_store'], extras['repository'])
                 txn_details = get_txn_id_from_store(repo_path, txn_id)
                 port = txn_details.get('port', 0)
 
             callback_daemon = HttpHooksCallbackDaemon(
                 txn_id=txn_id, host=host, port=port)
         case 'celery':
             from rhodecode.lib.hook_daemon.celery_hooks_deamon import CeleryHooksCallbackDaemon
-            callback_daemon = CeleryHooksCallbackDaemon(get_config(extras['config']))
+
+            config = get_app_config_lightweight(extras['config'])
+            task_queue = config.get('celery.broker_url')
+            task_backend = config.get('celery.result_backend')
+
+            callback_daemon = CeleryHooksCallbackDaemon(task_queue, task_backend)
         case 'local':
             from rhodecode.lib.hook_daemon.hook_module import Hooks
             callback_daemon = HooksModuleCallbackDaemon(Hooks.__module__)
         case _:
             log.error('Unsupported callback daemon protocol "%s"', protocol)
             raise Exception('Unsupported callback daemon protocol.')
 
     extras['hooks_uri'] = getattr(callback_daemon, 'hooks_uri', '')
     extras['task_queue'] = getattr(callback_daemon, 'task_queue', '')
     extras['task_backend'] = getattr(callback_daemon, 'task_backend', '')
     extras['hooks_protocol'] = protocol
     extras['time'] = time.time()
 
     # register txn_id
     extras['txn_id'] = txn_id
     log.debug('Prepared a callback daemon: %s',
               callback_daemon.__class__.__name__)
     return callback_daemon, extras

rhodecode/lib/hook_daemon/celery_hooks_deamon.py

@@ -1,33 +1,35 b''
 # Copyright (C) 2010-2023 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/
 
 from rhodecode.lib.hook_daemon.base import BaseHooksCallbackDaemon
 
 
 class CeleryHooksCallbackDaemon(BaseHooksCallbackDaemon):
     """
     Context manger for achieving a compatibility with celery backend
     """
 
-    def __init__(self, config):
-        # TODO: replace this with settings bootstrapped...
-        self.task_queue = config.get('app:main', 'celery.broker_url')
-        self.task_backend = config.get('app:main', 'celery.result_backend')
+    def __init__(self, task_queue, task_backend):
+        self.task_queue = task_queue
+        self.task_backend = task_backend
 
     def __repr__(self):
         return f'CeleryHooksCallbackDaemon(task_queue={self.task_queue}, task_backend={self.task_backend})'
+
+    def __repr__(self):
+        return f'CeleryHooksCallbackDaemon(task_queue={self.task_queue}, task_backend={self.task_backend})'

rhodecode/tests/fixture_mods/fixture_utils.py

@@ -1,1750 +1,1750 b''
 
 # Copyright (C) 2010-2023 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 os
 import re
 import pprint
 import shutil
 import socket
 import subprocess
 import time
 import uuid
 import dateutil.tz
 import logging
 import functools
 
 import mock
 import pyramid.testing
 import pytest
 import colander
 import requests
 import pyramid.paster
 
 import rhodecode
 import rhodecode.lib
 from rhodecode.model.changeset_status import ChangesetStatusModel
 from rhodecode.model.comment import CommentsModel
 from rhodecode.model.db import (
     PullRequest, PullRequestReviewers, 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.str_utils import safe_bytes
 from rhodecode.lib.hash_utils import sha1_safe
 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
 
 log = logging.getLogger(__name__)
 
 
 def cmp(a, b):
     # backport cmp from python2 so we can still use it in the custom code in this module
     return (a > b) - (a < b)
 
 
 @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.config_utils import get_app_config
     from rhodecode.config.middleware import make_pyramid_app
 
-    log.info("Using the RhodeCode configuration:{}".format(ini_config))
+    log.info("Using the RhodeCode configuration:%s", 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 = f'{backend_alias}-{dump_name}'
         backend = get_backend(backend_alias)
         dump_extractor = self.dump_extractors[backend_alias]
         repo_path = dump_extractor(dump_name, repo_name)
 
         vcs_repo = backend(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
     _master_repo_path = ''
     _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_id(self):
         # just fake some repo_id
         return self.repo.repo_id
 
     @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)
         self._master_repo_path = self._master_repo.repo_full_path
 
         return self._commit_ids
 
     def create_repo(
             self, commits=None, number_of_commits=0, heads=None,
             name_suffix='', 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': f'Commit {x} of {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, do_fetch=False):
         """
         Make sure that repo contains all commits mentioned in `heads`
         """
         vcsrepo = repo.scm_instance()
         vcsrepo.config.clear_section('hooks')
         commit_ids = [self._commit_ids[h] for h in heads]
         if do_fetch:
             vcsrepo.fetch(self._master_repo_path, commit_ids=commit_ids)
         vcsrepo.pull(self._master_repo_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=''):
         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 "%s_%s" % (
             self.invalid_repo_name.sub('_', self._test_name), len(self._cleanup_repos))
 
     def ensure_file(self, filename, content=b'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:
                 cleanup_message = message.replace(' ', '')
                 ref_name = f'refs/test-refs/{cleanup_message}'
                 refs[ref_name] = self._commit_ids[message]
             self._create_refs(repo, refs)
 
     def _create_refs(self, repo, refs):
         for ref_name, ref_val in refs.items():
             repo.set_refs(ref_name, ref_val)
 
 
 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()
 
     def __repr__(self):
         return f'{self.__class__.__name__}(alias={self.alias}, repo={self._repo_path})'
 
     @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 range(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 "{}_{}".format(
             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(safe_bytes(filename), content=safe_bytes(content)))
         imc.commit(
             message='Automatic commit from vcsbackend fixture',
             author='Automatic <[e-mail]>')
 
     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 vcsbackend_base(request, backend_alias, tests_tmp_path, baseapp, test_repo) -> VcsBackend:
     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 = f'vcs_test_{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
 
 
 def _add_commits_to_repo(vcs_repo, commits):
     commit_ids = {}
     if not commits:
         return commit_ids
 
     imc = vcs_repo.in_memory_commit
 
     for idx, commit in enumerate(commits):
         message = str(commit.get('message', f'Commit {idx}'))
 
         for node in commit.get('added', []):
             imc.add(FileNode(safe_bytes(node.path), content=node.content))
         for node in commit.get('changed', []):
             imc.change(FileNode(safe_bytes(node.path), content=node.content))
         for node in commit.get('removed', []):
             imc.remove(FileNode(safe_bytes(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(b'file_%b' % safe_bytes(str(idx)), content=safe_bytes(message)))
 
         commit = imc.commit(
             message=message,
             author=str(commit.get('author', 'Automatic <[e-mail]>')),
             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 = subprocess.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
     commit_ids: dict
 
     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='', reviewers=None, observers=None,
             title="Test", description="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(),
                 'observers': observers or self._get_observers(),
                 '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, branch: str = None) -> str:
         default_branch = branch or self.backend.default_branch_name
         message = self.commit_ids[commit_message]
         reference = f'branch:{default_branch}:{message}'
 
         return reference
 
     def _get_reviewers(self):
         role = PullRequestReviewers.ROLE_REVIEWER
         return [
             (TEST_USER_REGULAR_LOGIN, ['default1'], False, role, []),
             (TEST_USER_REGULAR2_LOGIN, ['default2'], False, role, []),
         ]
 
     def _get_observers(self):
         return [
 
         ]
 
     def update_source_repository(self, head=None, do_fetch=False):
         heads = [head or 'c3']
         self.backend.pull_heads(self.source_repository, heads=heads, do_fetch=do_fetch)
 
     def update_target_repository(self, head=None, do_fetch=False):
         heads = [head or 'c3']
         self.backend.pull_heads(self.target_repository, heads=heads, do_fetch=do_fetch)
 
     def set_pr_target_ref(self, ref_type: str = "branch", ref_name: str = "branch", ref_commit_id: str = "") -> str:
         full_ref = f"{ref_type}:{ref_name}:{ref_commit_id}"
         self.pull_request.target_ref = full_ref
         return full_ref
 
     def set_pr_source_ref(self, ref_type: str = "branch", ref_name: str = "branch", ref_commit_id: str = "") -> str:
         full_ref = f"{ref_type}:{ref_name}:{ref_commit_id}"
         self.pull_request.source_ref = full_ref
         return full_ref
 
     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, self.pull_request.author)
         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, self.pull_request.author)
         assert len(self.pull_request.revisions) == 1
         return removed_commit_id
 
     def create_comment(self, linked_to=None):
         comment = CommentsModel().create(
             text="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='n1', file_path='file_1'):
         comment = CommentsModel().create(
             text="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, key=functools.cmp_to_key(_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, key=functools.cmp_to_key(_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)
 
 
 @pytest.fixture(scope='session')
 def testrun():
     return {
         'uuid': uuid.uuid4(),
         'start': datetime.datetime.utcnow().isoformat(),
         'timestamp': int(time.time()),
     }
 
 
 class AppenlightClient(object):
 
     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 sha1_safe(f'{section}{value}{repo.repo_id}')
 
         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 sha1_safe(f'{section}{value}')
 
         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

rhodecode/tests/lib/test_hooks_daemon.py

@@ -1,363 +1,364 b''
 
 # Copyright (C) 2010-2023 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 logging
 import io
 
 import mock
 import msgpack
 import pytest
 import tempfile
 
 from rhodecode.lib.hook_daemon import http_hooks_deamon
 from rhodecode.lib.hook_daemon import celery_hooks_deamon
 from rhodecode.lib.hook_daemon import hook_module
 from rhodecode.lib.hook_daemon import base as hook_base
 from rhodecode.lib.str_utils import safe_bytes
 from rhodecode.tests.utils import assert_message_in_log
 from rhodecode.lib.ext_json import json
 
 test_proto = http_hooks_deamon.HooksHttpHandler.MSGPACK_HOOKS_PROTO
 
 
 class TestHooks(object):
     def test_hooks_can_be_used_as_a_context_processor(self):
         hooks = hook_module.Hooks()
         with hooks as return_value:
             pass
         assert hooks == return_value
 
 
 class TestHooksHttpHandler(object):
     def test_read_request_parses_method_name_and_arguments(self):
         data = {
             'method': 'test',
             'extras': {
                 'param1': 1,
                 'param2': 'a'
             }
         }
         request = self._generate_post_request(data)
         hooks_patcher = mock.patch.object(
             hook_module.Hooks, data['method'], create=True, return_value=1)
 
         with hooks_patcher as hooks_mock:
             handler = http_hooks_deamon.HooksHttpHandler
             handler.DEFAULT_HOOKS_PROTO = test_proto
             handler.wbufsize = 10240
             MockServer(handler, request)
 
         hooks_mock.assert_called_once_with(data['extras'])
 
     def test_hooks_serialized_result_is_returned(self):
         request = self._generate_post_request({})
         rpc_method = 'test'
         hook_result = {
             'first': 'one',
             'second': 2
         }
         extras = {}
 
         # patching our _read to return test method and proto used
         read_patcher = mock.patch.object(
             http_hooks_deamon.HooksHttpHandler, '_read_request',
             return_value=(test_proto, rpc_method, extras))
 
         # patch Hooks instance to return hook_result data on 'test' call
         hooks_patcher = mock.patch.object(
             hook_module.Hooks, rpc_method, create=True,
             return_value=hook_result)
 
         with read_patcher, hooks_patcher:
             handler = http_hooks_deamon.HooksHttpHandler
             handler.DEFAULT_HOOKS_PROTO = test_proto
             handler.wbufsize = 10240
             server = MockServer(handler, request)
 
         expected_result = http_hooks_deamon.HooksHttpHandler.serialize_data(hook_result)
 
         server.request.output_stream.seek(0)
         assert server.request.output_stream.readlines()[-1] == expected_result
 
     def test_exception_is_returned_in_response(self):
         request = self._generate_post_request({})
         rpc_method = 'test'
 
         read_patcher = mock.patch.object(
             http_hooks_deamon.HooksHttpHandler, '_read_request',
             return_value=(test_proto, rpc_method, {}))
 
         hooks_patcher = mock.patch.object(
             hook_module.Hooks, rpc_method, create=True,
             side_effect=Exception('Test exception'))
 
         with read_patcher, hooks_patcher:
             handler = http_hooks_deamon.HooksHttpHandler
             handler.DEFAULT_HOOKS_PROTO = test_proto
             handler.wbufsize = 10240
             server = MockServer(handler, request)
 
         server.request.output_stream.seek(0)
         data = server.request.output_stream.readlines()
         msgpack_data = b''.join(data[5:])
         org_exc = http_hooks_deamon.HooksHttpHandler.deserialize_data(msgpack_data)
         expected_result = {
             'exception': 'Exception',
             'exception_traceback': org_exc['exception_traceback'],
             'exception_args': ['Test exception']
         }
         assert org_exc == expected_result
 
     def test_log_message_writes_to_debug_log(self, caplog):
         ip_port = ('0.0.0.0', 8888)
         handler = http_hooks_deamon.HooksHttpHandler(MockRequest('POST /'), ip_port, mock.Mock())
         fake_date = '1/Nov/2015 00:00:00'
         date_patcher = mock.patch.object(
             handler, 'log_date_time_string', return_value=fake_date)
 
         with date_patcher, caplog.at_level(logging.DEBUG):
             handler.log_message('Some message %d, %s', 123, 'string')
 
         expected_message = f"HOOKS: client={ip_port} - - [{fake_date}] Some message 123, string"
 
         assert_message_in_log(
             caplog.records, expected_message,
             levelno=logging.DEBUG, module='http_hooks_deamon')
 
     def _generate_post_request(self, data, proto=test_proto):
         if proto == http_hooks_deamon.HooksHttpHandler.MSGPACK_HOOKS_PROTO:
             payload = msgpack.packb(data)
         else:
             payload = json.dumps(data)
 
         return b'POST / HTTP/1.0\nContent-Length: %d\n\n%b' % (
             len(payload), payload)
 
 
 class ThreadedHookCallbackDaemon(object):
     def test_constructor_calls_prepare(self):
         prepare_daemon_patcher = mock.patch.object(
             http_hooks_deamon.ThreadedHookCallbackDaemon, '_prepare')
         with prepare_daemon_patcher as prepare_daemon_mock:
             http_hooks_deamon.ThreadedHookCallbackDaemon()
         prepare_daemon_mock.assert_called_once_with()
 
     def test_run_is_called_on_context_start(self):
         patchers = mock.patch.multiple(
             http_hooks_deamon.ThreadedHookCallbackDaemon,
             _run=mock.DEFAULT, _prepare=mock.DEFAULT, __exit__=mock.DEFAULT)
 
         with patchers as mocks:
             daemon = http_hooks_deamon.ThreadedHookCallbackDaemon()
             with daemon as daemon_context:
                 pass
             mocks['_run'].assert_called_once_with()
         assert daemon_context == daemon
 
     def test_stop_is_called_on_context_exit(self):
         patchers = mock.patch.multiple(
             http_hooks_deamon.ThreadedHookCallbackDaemon,
             _run=mock.DEFAULT, _prepare=mock.DEFAULT, _stop=mock.DEFAULT)
 
         with patchers as mocks:
             daemon = http_hooks_deamon.ThreadedHookCallbackDaemon()
             with daemon as daemon_context:
                 assert mocks['_stop'].call_count == 0
 
             mocks['_stop'].assert_called_once_with()
         assert daemon_context == daemon
 
 
 class TestHttpHooksCallbackDaemon(object):
     def test_hooks_callback_generates_new_port(self, caplog):
         with caplog.at_level(logging.DEBUG):
             daemon = http_hooks_deamon.HttpHooksCallbackDaemon(host='127.0.0.1', port=8881)
         assert daemon._daemon.server_address == ('127.0.0.1', 8881)
 
         with caplog.at_level(logging.DEBUG):
             daemon = http_hooks_deamon.HttpHooksCallbackDaemon(host=None, port=None)
         assert daemon._daemon.server_address[1] in range(0, 66000)
         assert daemon._daemon.server_address[0] != '127.0.0.1'
 
     def test_prepare_inits_daemon_variable(self, tcp_server, caplog):
         with self._tcp_patcher(tcp_server), caplog.at_level(logging.DEBUG):
             daemon = http_hooks_deamon.HttpHooksCallbackDaemon(host='127.0.0.1', port=8881)
         assert daemon._daemon == tcp_server
 
         _, port = tcp_server.server_address
 
         msg = f"HOOKS: 127.0.0.1:{port}  Preparing HTTP callback daemon registering " \
               f"hook object: <class 'rhodecode.lib.hook_daemon.http_hooks_deamon.HooksHttpHandler'>"
         assert_message_in_log(
             caplog.records, msg, levelno=logging.DEBUG, module='http_hooks_deamon')
 
     def test_prepare_inits_hooks_uri_and_logs_it(
             self, tcp_server, caplog):
         with self._tcp_patcher(tcp_server), caplog.at_level(logging.DEBUG):
             daemon = http_hooks_deamon.HttpHooksCallbackDaemon(host='127.0.0.1', port=8881)
 
         _, port = tcp_server.server_address
         expected_uri = '{}:{}'.format('127.0.0.1', port)
         assert daemon.hooks_uri == expected_uri
 
         msg = f"HOOKS: 127.0.0.1:{port}  Preparing HTTP callback daemon registering " \
               f"hook object: <class 'rhodecode.lib.hook_daemon.http_hooks_deamon.HooksHttpHandler'>"
 
         assert_message_in_log(
             caplog.records, msg,
             levelno=logging.DEBUG, module='http_hooks_deamon')
 
     def test_run_creates_a_thread(self, tcp_server):
         thread = mock.Mock()
 
         with self._tcp_patcher(tcp_server):
             daemon = http_hooks_deamon.HttpHooksCallbackDaemon()
 
         with self._thread_patcher(thread) as thread_mock:
             daemon._run()
 
         thread_mock.assert_called_once_with(
             target=tcp_server.serve_forever,
             kwargs={'poll_interval': daemon.POLL_INTERVAL})
         assert thread.daemon is True
         thread.start.assert_called_once_with()
 
     def test_run_logs(self, tcp_server, caplog):
 
         with self._tcp_patcher(tcp_server):
             daemon = http_hooks_deamon.HttpHooksCallbackDaemon()
 
         with self._thread_patcher(mock.Mock()), caplog.at_level(logging.DEBUG):
             daemon._run()
 
         assert_message_in_log(
             caplog.records,
             'Running thread-based loop of callback daemon in background',
             levelno=logging.DEBUG, module='http_hooks_deamon')
 
     def test_stop_cleans_up_the_connection(self, tcp_server, caplog):
         thread = mock.Mock()
 
         with self._tcp_patcher(tcp_server):
             daemon = http_hooks_deamon.HttpHooksCallbackDaemon()
 
         with self._thread_patcher(thread), caplog.at_level(logging.DEBUG):
             with daemon:
                 assert daemon._daemon == tcp_server
                 assert daemon._callback_thread == thread
 
         assert daemon._daemon is None
         assert daemon._callback_thread is None
         tcp_server.shutdown.assert_called_with()
         thread.join.assert_called_once_with()
 
         assert_message_in_log(
             caplog.records, 'Waiting for background thread to finish.',
             levelno=logging.DEBUG, module='http_hooks_deamon')
 
     def _tcp_patcher(self, tcp_server):
         return mock.patch.object(
             http_hooks_deamon, 'TCPServer', return_value=tcp_server)
 
     def _thread_patcher(self, thread):
         return mock.patch.object(
             http_hooks_deamon.threading, 'Thread', return_value=thread)
 
 
 class TestPrepareHooksDaemon(object):
 
     @pytest.mark.parametrize('protocol', ('celery',))
     def test_returns_celery_hooks_callback_daemon_when_celery_protocol_specified(
             self, protocol):
         with tempfile.NamedTemporaryFile(mode='w') as temp_file:
             temp_file.write("[app:main]\ncelery.broker_url = redis://redis/0\n"
                             "celery.result_backend = redis://redis/0")
             temp_file.flush()
             expected_extras = {'config': temp_file.name}
             callback, extras = hook_base.prepare_callback_daemon(
                 expected_extras, protocol=protocol, host='')
             assert isinstance(callback, celery_hooks_deamon.CeleryHooksCallbackDaemon)
 
     @pytest.mark.parametrize('protocol, expected_class', (
         ('http', http_hooks_deamon.HttpHooksCallbackDaemon),
     ))
     def test_returns_real_hooks_callback_daemon_when_protocol_is_specified(
             self, protocol, expected_class):
         expected_extras = {
             'extra1': 'value1',
             'txn_id': 'txnid2',
             'hooks_protocol': protocol.lower(),
             'task_backend': '',
             'task_queue': '',
             'repo_store': '/var/opt/rhodecode_repo_store',
             'repository': 'rhodecode',
         }
         from rhodecode import CONFIG
         CONFIG['vcs.svn.redis_conn'] = 'redis://redis:6379/0'
         callback, extras = hook_base.prepare_callback_daemon(
             expected_extras.copy(), protocol=protocol, host='127.0.0.1',
             txn_id='txnid2')
         assert isinstance(callback, expected_class)
         extras.pop('hooks_uri')
         expected_extras['time'] = extras['time']
         assert extras == expected_extras
 
     @pytest.mark.parametrize('protocol', (
         'invalid',
         'Http',
         'HTTP',
+        'celerY'
     ))
     def test_raises_on_invalid_protocol(self, protocol):
         expected_extras = {
             'extra1': 'value1',
             'hooks_protocol': protocol.lower()
         }
         with pytest.raises(Exception):
             callback, extras = hook_base.prepare_callback_daemon(
                 expected_extras.copy(),
                 protocol=protocol, host='127.0.0.1')
 
 
 class MockRequest(object):
 
     def __init__(self, request):
         self.request = request
         self.input_stream = io.BytesIO(safe_bytes(self.request))
         self.output_stream = io.BytesIO()  # make it un-closable for testing invesitagion
         self.output_stream.close = lambda: None
 
     def makefile(self, mode, *args, **kwargs):
         return self.output_stream if mode == 'wb' else self.input_stream
 
 
 class MockServer(object):
 
     def __init__(self, handler_cls, request):
         ip_port = ('0.0.0.0', 8888)
         self.request = MockRequest(request)
         self.server_address = ip_port
         self.handler = handler_cls(self.request, ip_port, self)
 
 
 @pytest.fixture()
 def tcp_server():
     server = mock.Mock()
     server.server_address = ('127.0.0.1', 8881)
     server.wbufsize = 1024
     return server

rhodecode/tests/models/test_pullrequest.py

@@ -1,978 +1,979 b''
 
 # Copyright (C) 2010-2023 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 mock
 import pytest
 import textwrap
 
 import rhodecode
 from rhodecode.lib.vcs.backends import get_backend
 from rhodecode.lib.vcs.backends.base import (
     MergeResponse, MergeFailureReason, Reference)
 from rhodecode.lib.vcs.exceptions import RepositoryError
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.model.comment import CommentsModel
 from rhodecode.model.db import PullRequest, Session
 from rhodecode.model.pull_request import PullRequestModel
 from rhodecode.model.user import UserModel
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 from rhodecode.lib.str_utils import safe_str
 
 pytestmark = [
     pytest.mark.backends("git", "hg"),
 ]
 
 
 @pytest.mark.usefixtures('config_stub')
 class TestPullRequestModel(object):
 
     @pytest.fixture()
     def pull_request(self, request, backend, pr_util):
         """
         A pull request combined with multiples patches.
         """
         BackendClass = get_backend(backend.alias)
         merge_resp = MergeResponse(
             False, False, None, MergeFailureReason.UNKNOWN,
             metadata={'exception': 'MockError'})
         self.merge_patcher = mock.patch.object(
             BackendClass, 'merge', return_value=merge_resp)
         self.workspace_remove_patcher = mock.patch.object(
             BackendClass, 'cleanup_merge_workspace')
 
         self.workspace_remove_mock = self.workspace_remove_patcher.start()
         self.merge_mock = self.merge_patcher.start()
         self.comment_patcher = mock.patch(
             'rhodecode.model.changeset_status.ChangesetStatusModel.set_status')
         self.comment_patcher.start()
         self.notification_patcher = mock.patch(
             'rhodecode.model.notification.NotificationModel.create')
         self.notification_patcher.start()
         self.helper_patcher = mock.patch(
             'rhodecode.lib.helpers.route_path')
         self.helper_patcher.start()
 
         self.hook_patcher = mock.patch.object(PullRequestModel,
                                               'trigger_pull_request_hook')
         self.hook_mock = self.hook_patcher.start()
 
         self.invalidation_patcher = mock.patch(
             'rhodecode.model.pull_request.ScmModel.mark_for_invalidation')
         self.invalidation_mock = self.invalidation_patcher.start()
 
         self.pull_request = pr_util.create_pull_request(
             mergeable=True, name_suffix=u'ąć')
         self.source_commit = self.pull_request.source_ref_parts.commit_id
         self.target_commit = self.pull_request.target_ref_parts.commit_id
         self.workspace_id = 'pr-%s' % self.pull_request.pull_request_id
         self.repo_id = self.pull_request.target_repo.repo_id
 
         @request.addfinalizer
         def cleanup_pull_request():
             calls = [mock.call(
                 self.pull_request, self.pull_request.author, 'create')]
             self.hook_mock.assert_has_calls(calls)
 
             self.workspace_remove_patcher.stop()
             self.merge_patcher.stop()
             self.comment_patcher.stop()
             self.notification_patcher.stop()
             self.helper_patcher.stop()
             self.hook_patcher.stop()
             self.invalidation_patcher.stop()
 
         return self.pull_request
 
     def test_get_all(self, pull_request):
         prs = PullRequestModel().get_all(pull_request.target_repo)
         assert isinstance(prs, list)
         assert len(prs) == 1
 
     def test_count_all(self, pull_request):
         pr_count = PullRequestModel().count_all(pull_request.target_repo)
         assert pr_count == 1
 
     def test_get_awaiting_review(self, pull_request):
         prs = PullRequestModel().get_awaiting_review(pull_request.target_repo)
         assert isinstance(prs, list)
         assert len(prs) == 1
 
     def test_count_awaiting_review(self, pull_request):
         pr_count = PullRequestModel().count_awaiting_review(
             pull_request.target_repo)
         assert pr_count == 1
 
     def test_get_awaiting_my_review(self, pull_request):
         PullRequestModel().update_reviewers(
             pull_request, [(pull_request.author, ['author'], False, 'reviewer', [])],
             pull_request.author)
         Session().commit()
 
         prs = PullRequestModel().get_awaiting_my_review(
             pull_request.target_repo.repo_name, user_id=pull_request.author.user_id)
         assert isinstance(prs, list)
         assert len(prs) == 1
 
     def test_count_awaiting_my_review(self, pull_request):
         PullRequestModel().update_reviewers(
             pull_request, [(pull_request.author, ['author'], False, 'reviewer', [])],
             pull_request.author)
         Session().commit()
 
         pr_count = PullRequestModel().count_awaiting_my_review(
             pull_request.target_repo.repo_name, user_id=pull_request.author.user_id)
         assert pr_count == 1
 
     def test_delete_calls_cleanup_merge(self, pull_request):
         repo_id = pull_request.target_repo.repo_id
         PullRequestModel().delete(pull_request, pull_request.author)
         Session().commit()
 
         self.workspace_remove_mock.assert_called_once_with(
             repo_id, self.workspace_id)
 
     def test_close_calls_cleanup_and_hook(self, pull_request):
         PullRequestModel().close_pull_request(
             pull_request, pull_request.author)
         Session().commit()
 
         repo_id = pull_request.target_repo.repo_id
 
         self.workspace_remove_mock.assert_called_once_with(
             repo_id, self.workspace_id)
         self.hook_mock.assert_called_with(
             self.pull_request, self.pull_request.author, 'close')
 
     def test_merge_status(self, pull_request):
         self.merge_mock.return_value = MergeResponse(
             True, False, None, MergeFailureReason.NONE)
 
         assert pull_request._last_merge_source_rev is None
         assert pull_request._last_merge_target_rev is None
         assert pull_request.last_merge_status is None
 
         merge_response, status, msg = PullRequestModel().merge_status(pull_request)
         assert status is True
         assert msg == 'This pull request can be automatically merged.'
         self.merge_mock.assert_called_with(
             self.repo_id, self.workspace_id,
             pull_request.target_ref_parts,
             pull_request.source_repo.scm_instance(),
             pull_request.source_ref_parts, dry_run=True,
             use_rebase=False, close_branch=False)
 
         assert pull_request._last_merge_source_rev == self.source_commit
         assert pull_request._last_merge_target_rev == self.target_commit
         assert pull_request.last_merge_status is MergeFailureReason.NONE
 
         self.merge_mock.reset_mock()
         merge_response, status, msg = PullRequestModel().merge_status(pull_request)
         assert status is True
         assert msg == 'This pull request can be automatically merged.'
         assert self.merge_mock.called is False
 
     def test_merge_status_known_failure(self, pull_request):
         self.merge_mock.return_value = MergeResponse(
             False, False, None, MergeFailureReason.MERGE_FAILED,
             metadata={'unresolved_files': 'file1'})
 
         assert pull_request._last_merge_source_rev is None
         assert pull_request._last_merge_target_rev is None
         assert pull_request.last_merge_status is None
 
         merge_response, status, msg = PullRequestModel().merge_status(pull_request)
         assert status is False
         assert msg == 'This pull request cannot be merged because of merge conflicts. file1'
         self.merge_mock.assert_called_with(
             self.repo_id, self.workspace_id,
             pull_request.target_ref_parts,
             pull_request.source_repo.scm_instance(),
             pull_request.source_ref_parts, dry_run=True,
             use_rebase=False, close_branch=False)
 
         assert pull_request._last_merge_source_rev == self.source_commit
         assert pull_request._last_merge_target_rev == self.target_commit
         assert pull_request.last_merge_status is MergeFailureReason.MERGE_FAILED
 
         self.merge_mock.reset_mock()
         merge_response, status, msg = PullRequestModel().merge_status(pull_request)
         assert status is False
         assert msg == 'This pull request cannot be merged because of merge conflicts. file1'
         assert self.merge_mock.called is False
 
     def test_merge_status_unknown_failure(self, pull_request):
         self.merge_mock.return_value = MergeResponse(
             False, False, None, MergeFailureReason.UNKNOWN,
             metadata={'exception': 'MockError'})
 
         assert pull_request._last_merge_source_rev is None
         assert pull_request._last_merge_target_rev is None
         assert pull_request.last_merge_status is None
 
         merge_response, status, msg = PullRequestModel().merge_status(pull_request)
         assert status is False
         assert msg == (
             'This pull request cannot be merged because of an unhandled exception. '
             'MockError')
         self.merge_mock.assert_called_with(
             self.repo_id, self.workspace_id,
             pull_request.target_ref_parts,
             pull_request.source_repo.scm_instance(),
             pull_request.source_ref_parts, dry_run=True,
             use_rebase=False, close_branch=False)
 
         assert pull_request._last_merge_source_rev is None
         assert pull_request._last_merge_target_rev is None
         assert pull_request.last_merge_status is None
 
         self.merge_mock.reset_mock()
         merge_response, status, msg = PullRequestModel().merge_status(pull_request)
         assert status is False
         assert msg == (
             'This pull request cannot be merged because of an unhandled exception. '
             'MockError')
         assert self.merge_mock.called is True
 
     def test_merge_status_when_target_is_locked(self, pull_request):
         pull_request.target_repo.locked = [1, u'12345.50', 'lock_web']
         merge_response, status, msg = PullRequestModel().merge_status(pull_request)
         assert status is False
         assert msg == (
             'This pull request cannot be merged because the target repository '
             'is locked by user:1.')
 
     def test_merge_status_requirements_check_target(self, pull_request):
 
         def has_largefiles(self, repo):
             return repo == pull_request.source_repo
 
         patcher = mock.patch.object(PullRequestModel, '_has_largefiles', has_largefiles)
         with patcher:
             merge_response, status, msg = PullRequestModel().merge_status(pull_request)
 
         assert status is False
         assert msg == 'Target repository large files support is disabled.'
 
     def test_merge_status_requirements_check_source(self, pull_request):
 
         def has_largefiles(self, repo):
             return repo == pull_request.target_repo
 
         patcher = mock.patch.object(PullRequestModel, '_has_largefiles', has_largefiles)
         with patcher:
             merge_response, status, msg = PullRequestModel().merge_status(pull_request)
 
         assert status is False
         assert msg == 'Source repository large files support is disabled.'
 
     def test_merge(self, pull_request, merge_extras):
         user = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         merge_ref = Reference(
             'type', 'name', '6126b7bfcc82ad2d3deaee22af926b082ce54cc6')
         self.merge_mock.return_value = MergeResponse(
             True, True, merge_ref, MergeFailureReason.NONE)
 
         merge_extras['repository'] = pull_request.target_repo.repo_name
         PullRequestModel().merge_repo(
             pull_request, pull_request.author, extras=merge_extras)
         Session().commit()
 
         message = (
             u'Merge pull request !{pr_id} from {source_repo} {source_ref_name}'
             u'\n\n {pr_title}'.format(
                 pr_id=pull_request.pull_request_id,
                 source_repo=safe_str(
                     pull_request.source_repo.scm_instance().name),
                 source_ref_name=pull_request.source_ref_parts.name,
                 pr_title=safe_str(pull_request.title)
             )
         )
         self.merge_mock.assert_called_with(
             self.repo_id, self.workspace_id,
             pull_request.target_ref_parts,
             pull_request.source_repo.scm_instance(),
             pull_request.source_ref_parts,
             user_name=user.short_contact, user_email=user.email, message=message,
             use_rebase=False, close_branch=False
         )
         self.invalidation_mock.assert_called_once_with(
             pull_request.target_repo.repo_name)
 
         self.hook_mock.assert_called_with(
             self.pull_request, self.pull_request.author, 'merge')
 
         pull_request = PullRequest.get(pull_request.pull_request_id)
         assert pull_request.merge_rev == '6126b7bfcc82ad2d3deaee22af926b082ce54cc6'
 
     def test_merge_with_status_lock(self, pull_request, merge_extras):
         user = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         merge_ref = Reference(
             'type', 'name', '6126b7bfcc82ad2d3deaee22af926b082ce54cc6')
         self.merge_mock.return_value = MergeResponse(
             True, True, merge_ref, MergeFailureReason.NONE)
 
         merge_extras['repository'] = pull_request.target_repo.repo_name
 
         with pull_request.set_state(PullRequest.STATE_UPDATING):
             assert pull_request.pull_request_state == PullRequest.STATE_UPDATING
             PullRequestModel().merge_repo(
                 pull_request, pull_request.author, extras=merge_extras)
             Session().commit()
 
         assert pull_request.pull_request_state == PullRequest.STATE_CREATED
 
         message = (
             u'Merge pull request !{pr_id} from {source_repo} {source_ref_name}'
             u'\n\n {pr_title}'.format(
                 pr_id=pull_request.pull_request_id,
                 source_repo=safe_str(
                     pull_request.source_repo.scm_instance().name),
                 source_ref_name=pull_request.source_ref_parts.name,
                 pr_title=safe_str(pull_request.title)
             )
         )
         self.merge_mock.assert_called_with(
             self.repo_id, self.workspace_id,
             pull_request.target_ref_parts,
             pull_request.source_repo.scm_instance(),
             pull_request.source_ref_parts,
             user_name=user.short_contact, user_email=user.email, message=message,
             use_rebase=False, close_branch=False
         )
         self.invalidation_mock.assert_called_once_with(
             pull_request.target_repo.repo_name)
 
         self.hook_mock.assert_called_with(
             self.pull_request, self.pull_request.author, 'merge')
 
         pull_request = PullRequest.get(pull_request.pull_request_id)
         assert pull_request.merge_rev == '6126b7bfcc82ad2d3deaee22af926b082ce54cc6'
 
     def test_merge_failed(self, pull_request, merge_extras):
         user = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         merge_ref = Reference(
             'type', 'name', '6126b7bfcc82ad2d3deaee22af926b082ce54cc6')
         self.merge_mock.return_value = MergeResponse(
             False, False, merge_ref, MergeFailureReason.MERGE_FAILED)
 
         merge_extras['repository'] = pull_request.target_repo.repo_name
         PullRequestModel().merge_repo(
             pull_request, pull_request.author, extras=merge_extras)
         Session().commit()
 
         message = (
             u'Merge pull request !{pr_id} from {source_repo} {source_ref_name}'
             u'\n\n {pr_title}'.format(
                 pr_id=pull_request.pull_request_id,
                 source_repo=safe_str(
                     pull_request.source_repo.scm_instance().name),
                 source_ref_name=pull_request.source_ref_parts.name,
                 pr_title=safe_str(pull_request.title)
             )
         )
         self.merge_mock.assert_called_with(
             self.repo_id, self.workspace_id,
             pull_request.target_ref_parts,
             pull_request.source_repo.scm_instance(),
             pull_request.source_ref_parts,
             user_name=user.short_contact, user_email=user.email, message=message,
             use_rebase=False, close_branch=False
         )
 
         pull_request = PullRequest.get(pull_request.pull_request_id)
         assert self.invalidation_mock.called is False
         assert pull_request.merge_rev is None
 
     def test_get_commit_ids(self, pull_request):
         # The PR has been not merged yet, so expect an exception
         with pytest.raises(ValueError):
             PullRequestModel()._get_commit_ids(pull_request)
 
         # Merge revision is in the revisions list
         pull_request.merge_rev = pull_request.revisions[0]
         commit_ids = PullRequestModel()._get_commit_ids(pull_request)
         assert commit_ids == pull_request.revisions
 
         # Merge revision is not in the revisions list
         pull_request.merge_rev = 'f000' * 10
         commit_ids = PullRequestModel()._get_commit_ids(pull_request)
         assert commit_ids == pull_request.revisions + [pull_request.merge_rev]
 
     def test_get_diff_from_pr_version(self, pull_request):
         source_repo = pull_request.source_repo
         source_ref_id = pull_request.source_ref_parts.commit_id
         target_ref_id = pull_request.target_ref_parts.commit_id
         diff = PullRequestModel()._get_diff_from_pr_or_version(
             source_repo, source_ref_id, target_ref_id,
             hide_whitespace_changes=False, diff_context=6)
         assert b'file_1' in diff.raw.tobytes()
 
     def test_generate_title_returns_unicode(self):
         title = PullRequestModel().generate_pullrequest_title(
             source='source-dummy',
             source_ref='source-ref-dummy',
             target='target-dummy',
         )
         assert type(title) == str
 
     @pytest.mark.parametrize('title, has_wip', [
         ('hello', False),
         ('hello wip', False),
         ('hello wip: xxx', False),
         ('[wip] hello', True),
         ('[wip] hello', True),
         ('wip: hello', True),
         ('wip hello', True),
 
     ])
     def test_wip_title_marker(self, pull_request, title, has_wip):
         pull_request.title = title
         assert pull_request.work_in_progress == has_wip
 
 
 @pytest.mark.usefixtures('config_stub')
 class TestIntegrationMerge(object):
     @pytest.mark.parametrize('extra_config', (
         {'vcs.hooks.protocol.v2': 'celery', 'vcs.hooks.direct_calls': False},
     ))
     def test_merge_triggers_push_hooks(
             self, pr_util, user_admin, capture_rcextensions, merge_extras,
             extra_config):
 
         pull_request = pr_util.create_pull_request(
             approved=True, mergeable=True)
         # TODO: johbo: Needed for sqlite, try to find an automatic way for it
         merge_extras['repository'] = pull_request.target_repo.repo_name
         Session().commit()
 
         with mock.patch.dict(rhodecode.CONFIG, extra_config, clear=False):
             merge_state = PullRequestModel().merge_repo(
                 pull_request, user_admin, extras=merge_extras)
             Session().commit()
 
         assert merge_state.executed
         assert '_pre_push_hook' in capture_rcextensions
         assert '_push_hook' in capture_rcextensions
 
     def test_merge_can_be_rejected_by_pre_push_hook(
             self, pr_util, user_admin, capture_rcextensions, merge_extras):
         pull_request = pr_util.create_pull_request(
             approved=True, mergeable=True)
         # TODO: johbo: Needed for sqlite, try to find an automatic way for it
         merge_extras['repository'] = pull_request.target_repo.repo_name
         Session().commit()
 
         with mock.patch('rhodecode.EXTENSIONS.PRE_PUSH_HOOK') as pre_pull:
             pre_pull.side_effect = RepositoryError("Disallow push!")
             merge_status = PullRequestModel().merge_repo(
                 pull_request, user_admin, extras=merge_extras)
             Session().commit()
 
         assert not merge_status.executed
         assert 'pre_push' not in capture_rcextensions
         assert 'post_push' not in capture_rcextensions
 
     def test_merge_fails_if_target_is_locked(
             self, pr_util, user_regular, merge_extras):
         pull_request = pr_util.create_pull_request(
             approved=True, mergeable=True)
         locked_by = [user_regular.user_id + 1, 12345.50, 'lock_web']
         pull_request.target_repo.locked = locked_by
         # TODO: johbo: Check if this can work based on the database, currently
         # all data is pre-computed, that's why just updating the DB is not
         # enough.
         merge_extras['locked_by'] = locked_by
         merge_extras['repository'] = pull_request.target_repo.repo_name
         # TODO: johbo: Needed for sqlite, try to find an automatic way for it
         Session().commit()
         merge_status = PullRequestModel().merge_repo(
             pull_request, user_regular, extras=merge_extras)
         Session().commit()
 
         assert not merge_status.executed
 
 
 @pytest.mark.parametrize('use_outdated, inlines_count, outdated_count', [
     (False, 1, 0),
     (True, 0, 1),
 ])
 def test_outdated_comments(
         pr_util, use_outdated, inlines_count, outdated_count, config_stub):
     pull_request = pr_util.create_pull_request()
     pr_util.create_inline_comment(file_path='not_in_updated_diff')
 
     with outdated_comments_patcher(use_outdated) as outdated_comment_mock:
         pr_util.add_one_commit()
         assert_inline_comments(
             pull_request, visible=inlines_count, outdated=outdated_count)
     outdated_comment_mock.assert_called_with(pull_request)
 
 
 @pytest.mark.parametrize('mr_type, expected_msg', [
     (MergeFailureReason.NONE,
      'This pull request can be automatically merged.'),
     (MergeFailureReason.UNKNOWN,
      'This pull request cannot be merged because of an unhandled exception. CRASH'),
     (MergeFailureReason.MERGE_FAILED,
      'This pull request cannot be merged because of merge conflicts. CONFLICT_FILE'),
     (MergeFailureReason.PUSH_FAILED,
      'This pull request could not be merged because push to target:`some-repo@merge_commit` failed.'),
     (MergeFailureReason.TARGET_IS_NOT_HEAD,
      'This pull request cannot be merged because the target `ref_name` is not a head.'),
     (MergeFailureReason.HG_SOURCE_HAS_MORE_BRANCHES,
      'This pull request cannot be merged because the source contains more branches than the target.'),
     (MergeFailureReason.HG_TARGET_HAS_MULTIPLE_HEADS,
      'This pull request cannot be merged because the target `ref_name` has multiple heads: `a,b,c`.'),
     (MergeFailureReason.TARGET_IS_LOCKED,
      'This pull request cannot be merged because the target repository is locked by user:123.'),
     (MergeFailureReason.MISSING_TARGET_REF,
      'This pull request cannot be merged because the target reference `ref_name` is missing.'),
     (MergeFailureReason.MISSING_SOURCE_REF,
      'This pull request cannot be merged because the source reference `ref_name` is missing.'),
     (MergeFailureReason.SUBREPO_MERGE_FAILED,
      'This pull request cannot be merged because of conflicts related to sub repositories.'),
 
 ])
 def test_merge_response_message(mr_type, expected_msg):
     merge_ref = Reference('type', 'ref_name', '6126b7bfcc82ad2d3deaee22af926b082ce54cc6')
     metadata = {
         'unresolved_files': 'CONFLICT_FILE',
         'exception': "CRASH",
         'target': 'some-repo',
         'merge_commit': 'merge_commit',
         'target_ref': merge_ref,
         'source_ref': merge_ref,
         'heads': ','.join(['a', 'b', 'c']),
         'locked_by': 'user:123'
     }
 
     merge_response = MergeResponse(True, True, merge_ref, mr_type, metadata=metadata)
     assert merge_response.merge_status_message == expected_msg
 
 
 @pytest.fixture()
-def merge_extras(user_regular):
+def merge_extras(request, user_regular):
     """
     Context for the vcs operation when running a merge.
     """
+
     extras = {
         'ip': '127.0.0.1',
         'username': user_regular.username,
         'user_id': user_regular.user_id,
         'action': 'push',
         'repository': 'fake_target_repo_name',
         'scm': 'git',
-        'config': 'fake_config_ini_path',
+        'config': request.config.getini('pyramid_config'),
         'repo_store': '',
         'make_lock': None,
         'locked_by': [None, None, None],
         'server_url': 'http://test.example.com:5000',
         'hooks': ['push', 'pull'],
         'is_shadow_repo': False,
     }
     return extras
 
 
 @pytest.mark.usefixtures('config_stub')
 class TestUpdateCommentHandling(object):
 
     @pytest.fixture(autouse=True, scope='class')
     def enable_outdated_comments(self, request, baseapp):
         config_patch = mock.patch.dict(
             'rhodecode.CONFIG', {'rhodecode_use_outdated_comments': True})
         config_patch.start()
 
         @request.addfinalizer
         def cleanup():
             config_patch.stop()
 
     def test_comment_stays_unflagged_on_unchanged_diff(self, pr_util):
         commits = [
             {'message': 'a'},
             {'message': 'b', 'added': [FileNode(b'file_b', b'test_content\n')]},
             {'message': 'c', 'added': [FileNode(b'file_c', b'test_content\n')]},
         ]
         pull_request = pr_util.create_pull_request(
             commits=commits, target_head='a', source_head='b', revisions=['b'])
         pr_util.create_inline_comment(file_path='file_b')
         pr_util.add_one_commit(head='c')
 
         assert_inline_comments(pull_request, visible=1, outdated=0)
 
     def test_comment_stays_unflagged_on_change_above(self, pr_util):
         original_content = b''.join((b'line %d\n' % x for x in range(1, 11)))
         updated_content = b'new_line_at_top\n' + original_content
         commits = [
             {'message': 'a'},
             {'message': 'b', 'added': [FileNode(b'file_b', original_content)]},
             {'message': 'c', 'changed': [FileNode(b'file_b', updated_content)]},
         ]
         pull_request = pr_util.create_pull_request(
             commits=commits, target_head='a', source_head='b', revisions=['b'])
 
         with outdated_comments_patcher():
             comment = pr_util.create_inline_comment(
                 line_no=u'n8', file_path='file_b')
             pr_util.add_one_commit(head='c')
 
         assert_inline_comments(pull_request, visible=1, outdated=0)
         assert comment.line_no == u'n9'
 
     def test_comment_stays_unflagged_on_change_below(self, pr_util):
         original_content = b''.join([b'line %d\n' % x for x in range(10)])
         updated_content = original_content + b'new_line_at_end\n'
         commits = [
             {'message': 'a'},
             {'message': 'b', 'added': [FileNode(b'file_b', original_content)]},
             {'message': 'c', 'changed': [FileNode(b'file_b', updated_content)]},
         ]
         pull_request = pr_util.create_pull_request(
             commits=commits, target_head='a', source_head='b', revisions=['b'])
         pr_util.create_inline_comment(file_path='file_b')
         pr_util.add_one_commit(head='c')
 
         assert_inline_comments(pull_request, visible=1, outdated=0)
 
     @pytest.mark.parametrize('line_no', ['n4', 'o4', 'n10', 'o9'])
     def test_comment_flagged_on_change_around_context(self, pr_util, line_no):
         base_lines = [b'line %d\n' % x for x in range(1, 13)]
         change_lines = list(base_lines)
         change_lines.insert(6, b'line 6a added\n')
 
         # Changes on the last line of sight
         update_lines = list(change_lines)
         update_lines[0] = b'line 1 changed\n'
         update_lines[-1] = b'line 12 changed\n'
 
         def file_b(lines):
             return FileNode(b'file_b', b''.join(lines))
 
         commits = [
             {'message': 'a', 'added': [file_b(base_lines)]},
             {'message': 'b', 'changed': [file_b(change_lines)]},
             {'message': 'c', 'changed': [file_b(update_lines)]},
         ]
 
         pull_request = pr_util.create_pull_request(
             commits=commits, target_head='a', source_head='b', revisions=['b'])
         pr_util.create_inline_comment(line_no=line_no, file_path='file_b')
 
         with outdated_comments_patcher():
             pr_util.add_one_commit(head='c')
             assert_inline_comments(pull_request, visible=0, outdated=1)
 
     @pytest.mark.parametrize("change, content", [
         ('changed', b'changed\n'),
         ('removed', b''),
     ], ids=['changed', b'removed'])
     def test_comment_flagged_on_change(self, pr_util, change, content):
         commits = [
             {'message': 'a'},
             {'message': 'b', 'added': [FileNode(b'file_b', b'test_content\n')]},
             {'message': 'c', change: [FileNode(b'file_b', content)]},
         ]
         pull_request = pr_util.create_pull_request(
             commits=commits, target_head='a', source_head='b', revisions=['b'])
         pr_util.create_inline_comment(file_path='file_b')
 
         with outdated_comments_patcher():
             pr_util.add_one_commit(head='c')
             assert_inline_comments(pull_request, visible=0, outdated=1)
 
 
 @pytest.mark.usefixtures('config_stub')
 class TestUpdateChangedFiles(object):
 
     def test_no_changes_on_unchanged_diff(self, pr_util):
         commits = [
             {'message': 'a'},
             {'message': 'b',
              'added': [FileNode(b'file_b', b'test_content b\n')]},
             {'message': 'c',
              'added': [FileNode(b'file_c', b'test_content c\n')]},
         ]
         # open a PR from a to b, adding file_b
         pull_request = pr_util.create_pull_request(
             commits=commits, target_head='a', source_head='b', revisions=['b'],
             name_suffix='per-file-review')
 
         # modify PR adding new file file_c
         pr_util.add_one_commit(head='c')
 
         assert_pr_file_changes(
             pull_request,
             added=['file_c'],
             modified=[],
             removed=[])
 
     def test_modify_and_undo_modification_diff(self, pr_util):
         commits = [
             {'message': 'a'},
             {'message': 'b',
              'added': [FileNode(b'file_b', b'test_content b\n')]},
             {'message': 'c',
              'changed': [FileNode(b'file_b', b'test_content b modified\n')]},
             {'message': 'd',
              'changed': [FileNode(b'file_b', b'test_content b\n')]},
         ]
         # open a PR from a to b, adding file_b
         pull_request = pr_util.create_pull_request(
             commits=commits, target_head='a', source_head='b', revisions=['b'],
             name_suffix='per-file-review')
 
         # modify PR modifying file file_b
         pr_util.add_one_commit(head='c')
 
         assert_pr_file_changes(
             pull_request,
             added=[],
             modified=['file_b'],
             removed=[])
 
         # move the head again to d, which rollbacks change,
         # meaning we should indicate no changes
         pr_util.add_one_commit(head='d')
 
         assert_pr_file_changes(
             pull_request,
             added=[],
             modified=[],
             removed=[])
 
     def test_updated_all_files_in_pr(self, pr_util):
         commits = [
             {'message': 'a'},
             {'message': 'b', 'added': [
                 FileNode(b'file_a', b'test_content a\n'),
                 FileNode(b'file_b', b'test_content b\n'),
                 FileNode(b'file_c', b'test_content c\n')]},
             {'message': 'c', 'changed': [
                 FileNode(b'file_a', b'test_content a changed\n'),
                 FileNode(b'file_b', b'test_content b changed\n'),
                 FileNode(b'file_c', b'test_content c changed\n')]},
         ]
         # open a PR from a to b, changing 3 files
         pull_request = pr_util.create_pull_request(
             commits=commits, target_head='a', source_head='b', revisions=['b'],
             name_suffix='per-file-review')
 
         pr_util.add_one_commit(head='c')
 
         assert_pr_file_changes(
             pull_request,
             added=[],
             modified=['file_a', 'file_b', 'file_c'],
             removed=[])
 
     def test_updated_and_removed_all_files_in_pr(self, pr_util):
         commits = [
             {'message': 'a'},
             {'message': 'b', 'added': [
                 FileNode(b'file_a', b'test_content a\n'),
                 FileNode(b'file_b', b'test_content b\n'),
                 FileNode(b'file_c', b'test_content c\n')]},
             {'message': 'c', 'removed': [
                 FileNode(b'file_a', b'test_content a changed\n'),
                 FileNode(b'file_b', b'test_content b changed\n'),
                 FileNode(b'file_c', b'test_content c changed\n')]},
         ]
         # open a PR from a to b, removing 3 files
         pull_request = pr_util.create_pull_request(
             commits=commits, target_head='a', source_head='b', revisions=['b'],
             name_suffix='per-file-review')
 
         pr_util.add_one_commit(head='c')
 
         assert_pr_file_changes(
             pull_request,
             added=[],
             modified=[],
             removed=['file_a', 'file_b', 'file_c'])
 
 
 def test_update_writes_snapshot_into_pull_request_version(pr_util, config_stub):
     model = PullRequestModel()
     pull_request = pr_util.create_pull_request()
     pr_util.update_source_repository()
 
     model.update_commits(pull_request, pull_request.author)
 
     # Expect that it has a version entry now
     assert len(model.get_versions(pull_request)) == 1
 
 
 def test_update_skips_new_version_if_unchanged(pr_util, config_stub):
     pull_request = pr_util.create_pull_request()
     model = PullRequestModel()
     model.update_commits(pull_request, pull_request.author)
 
     # Expect that it still has no versions
     assert len(model.get_versions(pull_request)) == 0
 
 
 def test_update_assigns_comments_to_the_new_version(pr_util, config_stub):
     model = PullRequestModel()
     pull_request = pr_util.create_pull_request()
     comment = pr_util.create_comment()
     pr_util.update_source_repository()
 
     model.update_commits(pull_request, pull_request.author)
 
     # Expect that the comment is linked to the pr version now
     assert comment.pull_request_version == model.get_versions(pull_request)[0]
 
 
 def test_update_adds_a_comment_to_the_pull_request_about_the_change(pr_util, config_stub):
     model = PullRequestModel()
     pull_request = pr_util.create_pull_request()
     pr_util.update_source_repository()
     pr_util.update_source_repository()
 
     update_response = model.update_commits(pull_request, pull_request.author)
 
     commit_id = update_response.common_ancestor_id
     # Expect to find a new comment about the change
     expected_message = textwrap.dedent(
         """\
         Pull request updated. Auto status change to |under_review|
 
         .. role:: added
         .. role:: removed
         .. parsed-literal::
 
           Changed commits:
             * :added:`1 added`
             * :removed:`0 removed`
 
           Changed files:
             * `A file_2 <#a_c-{}-92ed3b5f07b4>`_
 
         .. |under_review| replace:: *"Under Review"*"""
     ).format(commit_id[:12])
     pull_request_comments = sorted(
         pull_request.comments, key=lambda c: c.modified_at)
     update_comment = pull_request_comments[-1]
     assert update_comment.text == expected_message
 
 
 def test_create_version_from_snapshot_updates_attributes(pr_util, config_stub):
     pull_request = pr_util.create_pull_request()
 
     # Avoiding default values
     pull_request.status = PullRequest.STATUS_CLOSED
     pull_request._last_merge_source_rev = "0" * 40
     pull_request._last_merge_target_rev = "1" * 40
     pull_request.last_merge_status = 1
     pull_request.merge_rev = "2" * 40
 
     # Remember automatic values
     created_on = pull_request.created_on
     updated_on = pull_request.updated_on
 
     # Create a new version of the pull request
     version = PullRequestModel()._create_version_from_snapshot(pull_request)
 
     # Check attributes
     assert version.title == pr_util.create_parameters['title']
     assert version.description == pr_util.create_parameters['description']
     assert version.status == PullRequest.STATUS_CLOSED
 
     # versions get updated created_on
     assert version.created_on != created_on
 
     assert version.updated_on == updated_on
     assert version.user_id == pull_request.user_id
     assert version.revisions == pr_util.create_parameters['revisions']
     assert version.source_repo == pr_util.source_repository
     assert version.source_ref == pr_util.create_parameters['source_ref']
     assert version.target_repo == pr_util.target_repository
     assert version.target_ref == pr_util.create_parameters['target_ref']
     assert version._last_merge_source_rev == pull_request._last_merge_source_rev
     assert version._last_merge_target_rev == pull_request._last_merge_target_rev
     assert version.last_merge_status == pull_request.last_merge_status
     assert version.merge_rev == pull_request.merge_rev
     assert version.pull_request == pull_request
 
 
 def test_link_comments_to_version_only_updates_unlinked_comments(pr_util, config_stub):
     version1 = pr_util.create_version_of_pull_request()
     comment_linked = pr_util.create_comment(linked_to=version1)
     comment_unlinked = pr_util.create_comment()
     version2 = pr_util.create_version_of_pull_request()
 
     PullRequestModel()._link_comments_to_version(version2)
     Session().commit()
 
     # Expect that only the new comment is linked to version2
     assert (
         comment_unlinked.pull_request_version_id ==
         version2.pull_request_version_id)
     assert (
         comment_linked.pull_request_version_id ==
         version1.pull_request_version_id)
     assert (
         comment_unlinked.pull_request_version_id !=
         comment_linked.pull_request_version_id)
 
 
 def test_calculate_commits():
     old_ids = [1, 2, 3]
     new_ids = [1, 3, 4, 5]
     change = PullRequestModel()._calculate_commit_id_changes(old_ids, new_ids)
     assert change.added == [4, 5]
     assert change.common == [1, 3]
     assert change.removed == [2]
     assert change.total == [1, 3, 4, 5]
 
 
 def assert_inline_comments(pull_request, visible=None, outdated=None):
     if visible is not None:
         inline_comments = CommentsModel().get_inline_comments(
             pull_request.target_repo.repo_id, pull_request=pull_request)
         inline_cnt = len(CommentsModel().get_inline_comments_as_list(
             inline_comments))
         assert inline_cnt == visible
     if outdated is not None:
         outdated_comments = CommentsModel().get_outdated_comments(
             pull_request.target_repo.repo_id, pull_request)
         assert len(outdated_comments) == outdated
 
 
 def assert_pr_file_changes(
         pull_request, added=None, modified=None, removed=None):
     pr_versions = PullRequestModel().get_versions(pull_request)
     # always use first version, ie original PR to calculate changes
     pull_request_version = pr_versions[0]
     old_diff_data, new_diff_data = PullRequestModel()._generate_update_diffs(
         pull_request, pull_request_version)
     file_changes = PullRequestModel()._calculate_file_changes(
         old_diff_data, new_diff_data)
 
     assert added == file_changes.added, \
         'expected added:%s vs value:%s' % (added, file_changes.added)
     assert modified == file_changes.modified, \
         'expected modified:%s vs value:%s' % (modified, file_changes.modified)
     assert removed == file_changes.removed, \
         'expected removed:%s vs value:%s' % (removed, file_changes.removed)
 
 
 def outdated_comments_patcher(use_outdated=True):
     return mock.patch.object(
         CommentsModel, 'use_outdated_comments',
         return_value=use_outdated)