fix(tests): fixed the creation of non-linear commits creation in tests

- This addresses and makes a fix for  #5706

rhodecode/api/tests/test_create_pull_request.py

@@ -1,367 +1,367 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 pytest
 
 from rhodecode.model.db import User
 from rhodecode.model.pull_request import PullRequestModel
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.user import UserModel
 from rhodecode.tests import TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN
 from rhodecode.api.tests.utils import build_data, api_call, assert_error
 
 
 @pytest.mark.usefixtures("testuser_api", "app")
 class TestCreatePullRequestApi(object):
     finalizers = []
 
     def teardown_method(self, method):
         if self.finalizers:
             for finalizer in self.finalizers:
                 finalizer()
             self.finalizers = []
 
     def test_create_with_wrong_data(self):
         required_data = {
             'source_repo': 'tests/source_repo',
             'target_repo': 'tests/target_repo',
             'source_ref': 'branch:default:initial',
             'target_ref': 'branch:default:new-feature',
         }
         for key in required_data:
             data = required_data.copy()
             data.pop(key)
             id_, params = build_data(
                 self.apikey, 'create_pull_request', **data)
             response = api_call(self.app, params)
 
             expected = 'Missing non optional `{}` arg in JSON DATA'.format(key)
             assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     @pytest.mark.parametrize('source_ref', [
         'bookmarg:default:initial'
     ])
     def test_create_with_wrong_refs_data(self, backend, source_ref):
 
         data = self._prepare_data(backend)
         data['source_ref'] = source_ref
 
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
 
         response = api_call(self.app, params)
 
         expected = "Ref `{}` type is not allowed. " \
                    "Only:['bookmark', 'book', 'tag', 'branch'] " \
                    "are possible.".format(source_ref)
         assert_error(id_, expected, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_create_with_correct_data(self, backend):
         data = self._prepare_data(backend)
         RepoModel().revoke_user_permission(
             self.source.repo_name, User.DEFAULT_USER)
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = "Created new pull request `{title}`".format(
             title=data['title'])
         result = response.json
         assert result['error'] is None
         assert result['result']['msg'] == expected_message
         pull_request_id = result['result']['pull_request_id']
         pull_request = PullRequestModel().get(pull_request_id)
         assert pull_request.title == data['title']
         assert pull_request.description == data['description']
         assert pull_request.source_ref == data['source_ref']
         assert pull_request.target_ref == data['target_ref']
         assert pull_request.source_repo.repo_name == data['source_repo']
         assert pull_request.target_repo.repo_name == data['target_repo']
         assert pull_request.revisions == [self.commit_ids['change']]
         assert len(pull_request.reviewers) == 1
 
     @pytest.mark.backends("git", "hg")
     def test_create_with_empty_description(self, backend):
         data = self._prepare_data(backend)
         data.pop('description')
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = "Created new pull request `{title}`".format(
             title=data['title'])
         result = response.json
         assert result['error'] is None
         assert result['result']['msg'] == expected_message
         pull_request_id = result['result']['pull_request_id']
         pull_request = PullRequestModel().get(pull_request_id)
         assert pull_request.description == ''
 
     @pytest.mark.backends("git", "hg")
     def test_create_with_empty_title(self, backend):
         data = self._prepare_data(backend)
         data.pop('title')
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         result = response.json
         pull_request_id = result['result']['pull_request_id']
         pull_request = PullRequestModel().get(pull_request_id)
         data['ref'] = backend.default_branch_name
         title = '{source_repo}#{ref} to {target_repo}'.format(**data)
         assert pull_request.title == title
 
     @pytest.mark.backends("git", "hg")
     def test_create_with_reviewers_specified_by_names(
             self, backend, no_notifications):
         data = self._prepare_data(backend)
         reviewers = [
             {'username': TEST_USER_REGULAR_LOGIN,
              'reasons': ['{} added manually'.format(TEST_USER_REGULAR_LOGIN)]},
             {'username': TEST_USER_ADMIN_LOGIN,
              'reasons': ['{} added manually'.format(TEST_USER_ADMIN_LOGIN)],
              'mandatory': True},
         ]
         data['reviewers'] = reviewers
 
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
 
         expected_message = "Created new pull request `{title}`".format(
             title=data['title'])
         result = response.json
         assert result['error'] is None
         assert result['result']['msg'] == expected_message
         pull_request_id = result['result']['pull_request_id']
         pull_request = PullRequestModel().get(pull_request_id)
 
         actual_reviewers = []
         for rev in pull_request.reviewers:
             entry = {
                 'username': rev.user.username,
                 'reasons': rev.reasons,
              }
             if rev.mandatory:
                 entry['mandatory'] = rev.mandatory
             actual_reviewers.append(entry)
 
         owner_username = pull_request.target_repo.user.username
         for spec_reviewer in reviewers[::]:
             # default reviewer will be added who is an owner of the repo
             # this get's overridden by a add owner to reviewers rule
             if spec_reviewer['username'] == owner_username:
                 spec_reviewer['reasons'] = [u'Default reviewer', u'Repository owner']
                 # since owner is more important, we don't inherit mandatory flag
                 del spec_reviewer['mandatory']
 
         assert sorted(actual_reviewers, key=lambda e: e['username']) \
                == sorted(reviewers, key=lambda e: e['username'])
 
     @pytest.mark.backends("git", "hg")
     def test_create_with_reviewers_specified_by_ids(
             self, backend, no_notifications):
         data = self._prepare_data(backend)
         reviewers = [
             {'username': UserModel().get_by_username(
                 TEST_USER_REGULAR_LOGIN).user_id,
              'reasons': ['added manually']},
             {'username': UserModel().get_by_username(
                 TEST_USER_ADMIN_LOGIN).user_id,
              'reasons': ['added manually']},
         ]
 
         data['reviewers'] = reviewers
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
 
         expected_message = "Created new pull request `{title}`".format(
             title=data['title'])
         result = response.json
         assert result['error'] is None
         assert result['result']['msg'] == expected_message
         pull_request_id = result['result']['pull_request_id']
         pull_request = PullRequestModel().get(pull_request_id)
 
         actual_reviewers = []
         for rev in pull_request.reviewers:
             entry = {
                 'username': rev.user.user_id,
                 'reasons': rev.reasons,
              }
             if rev.mandatory:
                 entry['mandatory'] = rev.mandatory
             actual_reviewers.append(entry)
 
         owner_user_id = pull_request.target_repo.user.user_id
         for spec_reviewer in reviewers[::]:
             # default reviewer will be added who is an owner of the repo
             # this get's overridden by a add owner to reviewers rule
             if spec_reviewer['username'] == owner_user_id:
                 spec_reviewer['reasons'] = [u'Default reviewer', u'Repository owner']
 
         assert sorted(actual_reviewers, key=lambda e: e['username']) \
                == sorted(reviewers, key=lambda e: e['username'])
 
     @pytest.mark.backends("git", "hg")
     def test_create_fails_when_the_reviewer_is_not_found(self, backend):
         data = self._prepare_data(backend)
         data['reviewers'] = [{'username': 'somebody'}]
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = 'user `somebody` does not exist'
         assert_error(id_, expected_message, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_cannot_create_with_reviewers_in_wrong_format(self, backend):
         data = self._prepare_data(backend)
         reviewers = ','.join([TEST_USER_REGULAR_LOGIN, TEST_USER_ADMIN_LOGIN])
         data['reviewers'] = reviewers
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = {u'': '"test_regular,test_admin" is not iterable'}
         assert_error(id_, expected_message, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_create_with_no_commit_hashes(self, backend):
         data = self._prepare_data(backend)
         expected_source_ref = data['source_ref']
         expected_target_ref = data['target_ref']
         data['source_ref'] = 'branch:{}'.format(backend.default_branch_name)
         data['target_ref'] = 'branch:{}'.format(backend.default_branch_name)
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = "Created new pull request `{title}`".format(
             title=data['title'])
         result = response.json
         assert result['result']['msg'] == expected_message
         pull_request_id = result['result']['pull_request_id']
         pull_request = PullRequestModel().get(pull_request_id)
         assert pull_request.source_ref == expected_source_ref
         assert pull_request.target_ref == expected_target_ref
 
     @pytest.mark.backends("git", "hg")
     @pytest.mark.parametrize("data_key", ["source_repo", "target_repo"])
     def test_create_fails_with_wrong_repo(self, backend, data_key):
         repo_name = 'fake-repo'
         data = self._prepare_data(backend)
         data[data_key] = repo_name
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = 'repository `{}` does not exist'.format(repo_name)
         assert_error(id_, expected_message, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     @pytest.mark.parametrize("data_key", ["source_ref", "target_ref"])
     def test_create_fails_with_non_existing_branch(self, backend, data_key):
         branch_name = 'test-branch'
         data = self._prepare_data(backend)
         data[data_key] = "branch:{}".format(branch_name)
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = 'The specified value:{type}:`{name}` ' \
                            'does not exist, or is not allowed.'.format(type='branch',
                                                                        name=branch_name)
         assert_error(id_, expected_message, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     @pytest.mark.parametrize("data_key", ["source_ref", "target_ref"])
     def test_create_fails_with_ref_in_a_wrong_format(self, backend, data_key):
         data = self._prepare_data(backend)
         ref = 'stange-ref'
         data[data_key] = ref
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = (
             'Ref `{ref}` given in a wrong format. Please check the API'
             ' documentation for more details'.format(ref=ref))
         assert_error(id_, expected_message, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     @pytest.mark.parametrize("data_key", ["source_ref", "target_ref"])
     def test_create_fails_with_non_existing_ref(self, backend, data_key):
         commit_id = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa10'
         ref = self._get_full_ref(backend, commit_id)
         data = self._prepare_data(backend)
         data[data_key] = ref
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = 'Ref `{}` does not exist'.format(ref)
         assert_error(id_, expected_message, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_create_fails_when_no_revisions(self, backend):
         data = self._prepare_data(backend, source_head='initial')
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = 'no commits found for merge between specified references'
         assert_error(id_, expected_message, given=response.body)
 
     @pytest.mark.backends("git", "hg")
     def test_create_fails_when_no_permissions(self, backend):
         data = self._prepare_data(backend)
         RepoModel().revoke_user_permission(
             self.source.repo_name, self.test_user)
         RepoModel().revoke_user_permission(
             self.source.repo_name, User.DEFAULT_USER)
 
         id_, params = build_data(
             self.apikey_regular, 'create_pull_request', **data)
         response = api_call(self.app, params)
         expected_message = 'repository `{}` does not exist'.format(
             self.source.repo_name)
         assert_error(id_, expected_message, given=response.body)
 
     def _prepare_data(
             self, backend, source_head='change', target_head='initial'):
         commits = [
             {'message': 'initial'},
             {'message': 'change'},
-            {'message': 'new-feature', 'parents': ['initial']},
+            {'message': 'new-feature', 'parents': ['initial'], 'branch': 'feature'},
         ]
         self.commit_ids = backend.create_master_repo(commits)
         self.source = backend.create_repo(heads=[source_head])
         self.target = backend.create_repo(heads=[target_head])
 
         data = {
             'source_repo': self.source.repo_name,
             'target_repo': self.target.repo_name,
             'source_ref': self._get_full_ref(
                 backend, self.commit_ids[source_head]),
             'target_ref': self._get_full_ref(
                 backend, self.commit_ids[target_head]),
             'title': 'Test PR 1',
             'description': 'Test'
         }
         RepoModel().grant_user_permission(
             self.source.repo_name, self.TEST_USER_LOGIN, 'repository.read')
         return data
 
     def _get_full_ref(self, backend, commit_id):
         return 'branch:{branch}:{commit_id}'.format(
             branch=backend.default_branch_name, commit_id=commit_id)

rhodecode/apps/repository/tests/test_repo_changelog.py

@@ -1,204 +1,204 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 re
 
 import pytest
 
 from rhodecode.apps.repository.views.repo_changelog import DEFAULT_CHANGELOG_SIZE
 from rhodecode.tests import TestController
 from rhodecode.tests.routes import route_path
 
 
 MATCH_HASH = re.compile(r'<span class="commit_hash">r(\d+):[\da-f]+</span>')
 
 
 def assert_commits_on_page(response, indexes):
     found_indexes = [int(idx) for idx in MATCH_HASH.findall(response.text)]
     assert found_indexes == indexes
 
 
 class TestChangelogController(TestController):
 
     def test_commits_page(self, backend):
         self.log_user()
         response = self.app.get(
             route_path('repo_commits', repo_name=backend.repo_name))
 
         first_idx = -1
         last_idx = -DEFAULT_CHANGELOG_SIZE
         self.assert_commit_range_on_page(response, first_idx, last_idx, backend)
 
     def test_changelog(self, backend):
         self.log_user()
         response = self.app.get(
             route_path('repo_changelog', repo_name=backend.repo_name))
 
         first_idx = -1
         last_idx = -DEFAULT_CHANGELOG_SIZE
         self.assert_commit_range_on_page(
             response, first_idx, last_idx, backend)
 
     @pytest.mark.backends("hg", "git")
     def test_changelog_filtered_by_branch(self, backend):
         self.log_user()
         self.app.get(
             route_path('repo_changelog', repo_name=backend.repo_name,
                        params=dict(branch=backend.default_branch_name)),
             status=200)
 
     @pytest.mark.backends("hg", "git")
     def test_commits_filtered_by_branch(self, backend):
         self.log_user()
         self.app.get(
             route_path('repo_commits', repo_name=backend.repo_name,
                        params=dict(branch=backend.default_branch_name)),
             status=200)
 
     @pytest.mark.backends("svn")
     def test_changelog_filtered_by_branch_svn(self, autologin_user, backend):
         repo = backend['svn-simple-layout']
         response = self.app.get(
             route_path('repo_changelog', repo_name=repo.repo_name,
                        params=dict(branch='trunk')),
             status=200)
 
         assert_commits_on_page(response, indexes=[15, 12, 7, 3, 2, 1])
 
     def test_commits_filtered_by_wrong_branch(self, backend):
         self.log_user()
         branch = 'wrong-branch-name'
         response = self.app.get(
             route_path('repo_commits', repo_name=backend.repo_name,
                        params=dict(branch=branch)),
             status=302)
         expected_url = '/{repo}/commits/{branch}'.format(
             repo=backend.repo_name, branch=branch)
         assert expected_url in response.location
         response = response.follow()
         expected_warning = f'Branch {branch} is not found.'
         assert expected_warning in response.text
 
     @pytest.mark.xfail_backends("svn", reason="Depends on branch support")
     def test_changelog_filtered_by_branch_with_merges(self, autologin_user, backend):
 
         # Note: The changelog of branch "b" does not contain the commit "a1"
         # although this is a parent of commit "b1". And branch "b" has commits
         # which have a smaller index than commit "a1".
         commits = [
-            {'message': 'a'},
+            {'message': 'a', 'branch': 'master'},
             {'message': 'b', 'branch': 'b'},
-            {'message': 'a1', 'parents': ['a']},
+            {'message': 'a1', 'parents': ['a'], 'branch': 'master'},
             {'message': 'b1', 'branch': 'b', 'parents': ['b', 'a1']},
         ]
         backend.create_repo(commits)
 
         self.app.get(
             route_path('repo_changelog', repo_name=backend.repo_name,
                        params=dict(branch='b')),
             status=200)
 
     @pytest.mark.backends("hg")
     def test_commits_closed_branches(self, autologin_user, backend):
         repo = backend['closed_branch']
         response = self.app.get(
             route_path('repo_commits', repo_name=repo.repo_name,
                        params=dict(branch='experimental')),
             status=200)
 
         assert_commits_on_page(response, indexes=[3, 1])
 
     def test_changelog_pagination(self, backend):
         self.log_user()
         # pagination, walk up to page 6
         changelog_url = route_path(
             'repo_commits', repo_name=backend.repo_name)
 
         for page in range(1, 7):
             response = self.app.get(changelog_url, {'page': page})
 
         first_idx = -DEFAULT_CHANGELOG_SIZE * (page - 1) - 1
         last_idx = -DEFAULT_CHANGELOG_SIZE * page
         self.assert_commit_range_on_page(response, first_idx, last_idx, backend)
 
     def assert_commit_range_on_page(
             self, response, first_idx, last_idx, backend):
         input_template = (
             """<input class="commit-range" """ 
             """data-commit-id="%(raw_id)s" data-commit-idx="%(idx)s" """ 
             """data-short-id="%(short_id)s" id="%(raw_id)s" """
             """name="%(raw_id)s" type="checkbox" value="1" />"""
         )
 
         commit_span_template = """<span class="commit_hash">r%s:%s</span>"""
         repo = backend.repo
 
         first_commit_on_page = repo.get_commit(commit_idx=first_idx)
         response.mustcontain(
             input_template % {'raw_id': first_commit_on_page.raw_id,
                               'idx': first_commit_on_page.idx,
                               'short_id': first_commit_on_page.short_id})
 
         response.mustcontain(commit_span_template % (
             first_commit_on_page.idx, first_commit_on_page.short_id)
         )
 
         last_commit_on_page = repo.get_commit(commit_idx=last_idx)
         response.mustcontain(
             input_template % {'raw_id': last_commit_on_page.raw_id,
                               'idx': last_commit_on_page.idx,
                               'short_id': last_commit_on_page.short_id})
         response.mustcontain(commit_span_template % (
             last_commit_on_page.idx, last_commit_on_page.short_id)
         )
 
         first_commit_of_next_page = repo.get_commit(commit_idx=last_idx - 1)
         first_span_of_next_page = commit_span_template % (
             first_commit_of_next_page.idx, first_commit_of_next_page.short_id)
         assert first_span_of_next_page not in response
 
     @pytest.mark.parametrize('test_path', [
         'vcs/exceptions.py',
         '/vcs/exceptions.py',
         '//vcs/exceptions.py'
     ])
     def test_commits_with_filenode(self, backend, test_path):
         self.log_user()
         response = self.app.get(
             route_path('repo_commits_file', repo_name=backend.repo_name,
                        commit_id='tip', f_path=test_path),
             )
 
         # history commits messages
         response.mustcontain('Added exceptions module, this time for real')
         response.mustcontain('Added not implemented hg backend test case')
         response.mustcontain('Added BaseChangeset class')
 
     def test_commits_with_filenode_that_is_dirnode(self, backend):
         self.log_user()
         self.app.get(
             route_path('repo_commits_file', repo_name=backend.repo_name,
                        commit_id='tip', f_path='/tests'),
             status=302)
 
     def test_commits_with_filenode_not_existing(self, backend):
         self.log_user()
         self.app.get(
             route_path('repo_commits_file', repo_name=backend.repo_name,
                        commit_id='tip', f_path='wrong_path'),
             status=302)

rhodecode/apps/repository/tests/test_repo_compare.py

@@ -1,656 +1,656 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 lxml.html
 
 from rhodecode.lib.vcs.exceptions import RepositoryRequirementError
 from rhodecode.tests import assert_session_flash
 from rhodecode.tests.utils import AssertResponse, commit_change
 from rhodecode.tests.routes import route_path
 
 
 @pytest.mark.usefixtures("autologin_user", "app")
 class TestCompareView(object):
 
     def test_compare_index_is_reached_at_least_once(self, backend):
         repo = backend.repo
         self.app.get(
             route_path('repo_compare_select', repo_name=repo.repo_name))
 
     @pytest.mark.xfail_backends("svn", reason="Requires pull")
     def test_compare_remote_with_different_commit_indexes(self, backend):
         # Preparing the following repository structure:
         #
         # Origin repository has two commits:
         #
         #    0    1
         #    A -- D
         #
         # The fork of it has a few more commits and "D" has a commit index
         # which does not exist in origin.
         #
         #    0    1    2    3    4
         #    A --   --   -- D -- E
         #      \- B -- C
         #
 
         fork = backend.create_repo()
         origin = backend.create_repo()
 
         # prepare fork
         commit0 = commit_change(
             fork.repo_name, filename=b'file1', content=b'A',
             message='A - Initial Commit', vcs_type=backend.alias, parent=None, newfile=True)
 
         commit1 = commit_change(
             fork.repo_name, filename=b'file1', content=b'B',
             message='B, child of A', vcs_type=backend.alias, parent=commit0)
 
         commit_change(  # commit 2
             fork.repo_name, filename=b'file1', content=b'C',
             message='C, child of B', vcs_type=backend.alias, parent=commit1)
 
         commit3 = commit_change(
             fork.repo_name, filename=b'file1', content=b'D',
-            message='D, child of A', vcs_type=backend.alias, parent=commit0)
+            message='D, child of A', vcs_type=backend.alias, parent=commit0, branch='feature')
 
         commit4 = commit_change(
             fork.repo_name, filename=b'file1', content=b'E',
-            message='E, child of D', vcs_type=backend.alias, parent=commit3)
+            message='E, child of D', vcs_type=backend.alias, parent=commit3, branch='feature')
 
         # prepare origin repository, taking just the history up to D
 
         origin_repo = origin.scm_instance(cache=False)
         origin_repo.config.clear_section('hooks')
         origin_repo.pull(fork.repo_full_path, commit_ids=[commit3.raw_id])
         origin_repo = origin.scm_instance(cache=False)  # cache rebuild
 
         # Verify test fixture setup
         # This does not work for git
         if backend.alias != 'git':
             assert 5 == len(fork.scm_instance(cache=False).commit_ids)
             assert 2 == len(origin_repo.commit_ids)
 
         # Comparing the revisions
         response = self.app.get(
             route_path('repo_compare',
                        repo_name=origin.repo_name,
                        source_ref_type="rev", source_ref=commit3.raw_id,
                        target_ref_type="rev", target_ref=commit4.raw_id,
                        params=dict(merge='1', target_repo=fork.repo_name)
                        ),
             status=200)
 
         compare_page = ComparePage(response)
         compare_page.contains_commits([commit4])
 
     @pytest.mark.xfail_backends("svn", reason="Depends on branch support")
     def test_compare_forks_on_branch_extra_commits(self, backend):
         repo1 = backend.create_repo()
 
         # commit something !
         commit0 = commit_change(
             repo1.repo_name, filename=b'file1', content=b'line1\n',
             message='commit1', vcs_type=backend.alias, parent=None,
             newfile=True)
 
         # fork this repo
         repo2 = backend.create_fork()
 
         # add two extra commit into fork
         commit1 = commit_change(
             repo2.repo_name, filename=b'file1', content=b'line1\nline2\n',
             message='commit2', vcs_type=backend.alias, parent=commit0)
 
         commit2 = commit_change(
             repo2.repo_name, filename=b'file1', content=b'line1\nline2\nline3\n',
             message='commit3', vcs_type=backend.alias, parent=commit1)
 
         commit_id1 = repo1.scm_instance().DEFAULT_BRANCH_NAME
         commit_id2 = repo2.scm_instance().DEFAULT_BRANCH_NAME
 
         response = self.app.get(
             route_path('repo_compare',
                        repo_name=repo1.repo_name,
                        source_ref_type="branch", source_ref=commit_id2,
                        target_ref_type="branch", target_ref=commit_id1,
                        params=dict(merge='1', target_repo=repo2.repo_name)
                        ))
 
         response.mustcontain('%s@%s' % (repo1.repo_name, commit_id2))
         response.mustcontain('%s@%s' % (repo2.repo_name, commit_id1))
 
         compare_page = ComparePage(response)
         compare_page.contains_change_summary(1, 2, 0)
         compare_page.contains_commits([commit1, commit2])
 
         anchor = 'a_c-{}-826e8142e6ba'.format(commit0.short_id)
         compare_page.contains_file_links_and_anchors([('file1', anchor), ])
 
         # Swap is removed when comparing branches since it's a PR feature and
         # it is then a preview mode
         compare_page.swap_is_hidden()
         compare_page.target_source_are_disabled()
 
     @pytest.mark.xfail_backends("svn", reason="Depends on branch support")
     def test_compare_forks_on_branch_extra_commits_origin_has_incoming(self, backend):
         repo1 = backend.create_repo()
 
         # commit something !
         commit0 = commit_change(
             repo1.repo_name, filename=b'file1', content=b'line1\n',
             message='commit1', vcs_type=backend.alias, parent=None,
             newfile=True)
 
         # fork this repo
         repo2 = backend.create_fork()
 
         # now commit something to origin repo
         commit_change(
             repo1.repo_name, filename=b'file2', content=b'line1file2\n',
             message='commit2', vcs_type=backend.alias, parent=commit0,
             newfile=True)
 
         # add two extra commit into fork
         commit1 = commit_change(
             repo2.repo_name, filename=b'file1', content=b'line1\nline2\n',
             message='commit2', vcs_type=backend.alias, parent=commit0)
 
         commit2 = commit_change(
             repo2.repo_name, filename=b'file1', content=b'line1\nline2\nline3\n',
             message='commit3', vcs_type=backend.alias, parent=commit1)
 
         commit_id1 = repo1.scm_instance().DEFAULT_BRANCH_NAME
         commit_id2 = repo2.scm_instance().DEFAULT_BRANCH_NAME
 
         response = self.app.get(
             route_path('repo_compare',
                        repo_name=repo1.repo_name,
                        source_ref_type="branch", source_ref=commit_id2,
                        target_ref_type="branch", target_ref=commit_id1,
                        params=dict(merge='1', target_repo=repo2.repo_name),
                        ))
 
         response.mustcontain(f'{repo1.repo_name}@{commit_id2}')
         response.mustcontain(f'{repo2.repo_name}@{commit_id1}')
 
         compare_page = ComparePage(response)
         compare_page.contains_change_summary(1, 2, 0)
         compare_page.contains_commits([commit1, commit2])
         anchor = f'a_c-{commit0.short_id}-826e8142e6ba'
         compare_page.contains_file_links_and_anchors([('file1', anchor), ])
 
         # Swap is removed when comparing branches since it's a PR feature and
         # it is then a preview mode
         compare_page.swap_is_hidden()
         compare_page.target_source_are_disabled()
 
     @pytest.mark.xfail_backends("svn")
     # TODO(marcink): no svn support for compare two seperate repos
     def test_compare_of_unrelated_forks(self, backend):
         orig = backend.create_repo(number_of_commits=1)
         fork = backend.create_repo(number_of_commits=1)
 
         response = self.app.get(
             route_path('repo_compare',
                        repo_name=orig.repo_name,
                        source_ref_type="rev", source_ref="tip",
                        target_ref_type="rev", target_ref="tip",
                        params=dict(merge='1', target_repo=fork.repo_name),
                        ),
             status=302)
         response = response.follow()
         response.mustcontain("Repositories unrelated.")
 
     @pytest.mark.xfail_backends("svn")
     def test_compare_cherry_pick_commits_from_bottom(self, backend):
 
         # repo1:
         #     commit0:
         #     commit1:
         # repo1-fork- in which we will cherry pick bottom commits
         #     commit0:
         #     commit1:
         #     commit2: x
         #     commit3: x
         #     commit4: x
         #     commit5:
         # make repo1, and commit1+commit2
 
         repo1 = backend.create_repo()
 
         # commit something !
         commit0 = commit_change(
             repo1.repo_name, filename=b'file1', content=b'line1\n',
             message='commit1', vcs_type=backend.alias, parent=None,
             newfile=True)
         commit1 = commit_change(
             repo1.repo_name, filename=b'file1', content=b'line1\nline2\n',
             message='commit2', vcs_type=backend.alias, parent=commit0)
 
         # fork this repo
         repo2 = backend.create_fork()
 
         # now make commit3-6
         commit2 = commit_change(
             repo1.repo_name, filename=b'file1', content=b'line1\nline2\nline3\n',
             message='commit3', vcs_type=backend.alias, parent=commit1)
         commit3 = commit_change(
             repo1.repo_name, filename=b'file1',content=b'line1\nline2\nline3\nline4\n',
             message='commit4', vcs_type=backend.alias, parent=commit2)
         commit4 = commit_change(
             repo1.repo_name, filename=b'file1', content=b'line1\nline2\nline3\nline4\nline5\n',
             message='commit5', vcs_type=backend.alias, parent=commit3)
         commit_change(  # commit 5
             repo1.repo_name, filename=b'file1', content=b'line1\nline2\nline3\nline4\nline5\nline6\n',
             message='commit6', vcs_type=backend.alias, parent=commit4)
 
         response = self.app.get(
             route_path('repo_compare',
                        repo_name=repo2.repo_name,
                        # parent of commit2, in target repo2
                        source_ref_type="rev", source_ref=commit1.raw_id,
                        target_ref_type="rev", target_ref=commit4.raw_id,
                        params=dict(merge='1', target_repo=repo1.repo_name),
                        ))
         response.mustcontain('%s@%s' % (repo2.repo_name, commit1.short_id))
         response.mustcontain('%s@%s' % (repo1.repo_name, commit4.short_id))
 
         # files
         compare_page = ComparePage(response)
         compare_page.contains_change_summary(1, 3, 0)
         compare_page.contains_commits([commit2, commit3, commit4])
         anchor = 'a_c-{}-826e8142e6ba'.format(commit1.short_id)
         compare_page.contains_file_links_and_anchors([('file1', anchor),])
 
     @pytest.mark.xfail_backends("svn")
     def test_compare_cherry_pick_commits_from_top(self, backend):
         # repo1:
         #     commit0:
         #     commit1:
         # repo1-fork- in which we will cherry pick bottom commits
         #     commit0:
         #     commit1:
         #     commit2:
         #     commit3: x
         #     commit4: x
         #     commit5: x
 
         # make repo1, and commit1+commit2
         repo1 = backend.create_repo()
 
         # commit something !
         commit0 = commit_change(
             repo1.repo_name, filename=b'file1', content=b'line1\n',
             message='commit1', vcs_type=backend.alias, parent=None,
             newfile=True)
         commit1 = commit_change(
             repo1.repo_name, filename=b'file1', content=b'line1\nline2\n',
             message='commit2', vcs_type=backend.alias, parent=commit0)
 
         # fork this repo
         backend.create_fork()
 
         # now make commit3-6
         commit2 = commit_change(
             repo1.repo_name, filename=b'file1', content=b'line1\nline2\nline3\n',
             message='commit3', vcs_type=backend.alias, parent=commit1)
         commit3 = commit_change(
             repo1.repo_name, filename=b'file1',
             content=b'line1\nline2\nline3\nline4\n', message='commit4',
             vcs_type=backend.alias, parent=commit2)
         commit4 = commit_change(
             repo1.repo_name, filename=b'file1',
             content=b'line1\nline2\nline3\nline4\nline5\n', message='commit5',
             vcs_type=backend.alias, parent=commit3)
         commit5 = commit_change(
             repo1.repo_name, filename=b'file1',
             content=b'line1\nline2\nline3\nline4\nline5\nline6\n',
             message='commit6', vcs_type=backend.alias, parent=commit4)
 
         response = self.app.get(
             route_path('repo_compare',
                        repo_name=repo1.repo_name,
                        # parent of commit3, not in source repo2
                        source_ref_type="rev", source_ref=commit2.raw_id,
                        target_ref_type="rev", target_ref=commit5.raw_id,
                        params=dict(merge='1'),))
 
         response.mustcontain('%s@%s' % (repo1.repo_name, commit2.short_id))
         response.mustcontain('%s@%s' % (repo1.repo_name, commit5.short_id))
 
         compare_page = ComparePage(response)
         compare_page.contains_change_summary(1, 3, 0)
         compare_page.contains_commits([commit3, commit4, commit5])
 
         # files
         anchor = 'a_c-{}-826e8142e6ba'.format(commit2.short_id)
         compare_page.contains_file_links_and_anchors([('file1', anchor),])
 
     @pytest.mark.xfail_backends("svn")
     def test_compare_remote_branches(self, backend):
         repo1 = backend.repo
         repo2 = backend.create_fork()
 
         commit_id1 = repo1.get_commit(commit_idx=3).raw_id
         commit_id1_short = repo1.get_commit(commit_idx=3).short_id
         commit_id2 = repo1.get_commit(commit_idx=6).raw_id
         commit_id2_short = repo1.get_commit(commit_idx=6).short_id
 
         response = self.app.get(
             route_path('repo_compare',
                        repo_name=repo1.repo_name,
                        source_ref_type="rev", source_ref=commit_id1,
                        target_ref_type="rev", target_ref=commit_id2,
                        params=dict(merge='1', target_repo=repo2.repo_name),
                        ))
 
         response.mustcontain('%s@%s' % (repo1.repo_name, commit_id1))
         response.mustcontain('%s@%s' % (repo2.repo_name, commit_id2))
 
         compare_page = ComparePage(response)
 
         # outgoing commits between those commits
         compare_page.contains_commits(
             [repo2.get_commit(commit_idx=x) for x in [4, 5, 6]])
 
         # files
         compare_page.contains_file_links_and_anchors([
             ('vcs/backends/hg.py', 'a_c-{}-9c390eb52cd6'.format(commit_id2_short)),
             ('vcs/backends/__init__.py', 'a_c-{}-41b41c1f2796'.format(commit_id1_short)),
             ('vcs/backends/base.py', 'a_c-{}-2f574d260608'.format(commit_id1_short)),
         ])
 
     @pytest.mark.xfail_backends("svn")
     def test_source_repo_new_commits_after_forking_simple_diff(self, backend):
         repo1 = backend.create_repo()
         r1_name = repo1.repo_name
 
         commit0 = commit_change(
             repo=r1_name, filename=b'file1',
             content=b'line1', message='commit1', vcs_type=backend.alias,
             newfile=True)
         assert repo1.scm_instance().commit_ids == [commit0.raw_id]
 
         # fork the repo1
         repo2 = backend.create_fork()
         assert repo2.scm_instance().commit_ids == [commit0.raw_id]
 
         self.r2_id = repo2.repo_id
         r2_name = repo2.repo_name
 
         commit1 = commit_change(
             repo=r2_name, filename=b'file1-fork',
             content=b'file1-line1-from-fork', message='commit1-fork',
             vcs_type=backend.alias, parent=repo2.scm_instance()[-1],
             newfile=True)
 
         commit2 = commit_change(
             repo=r2_name, filename=b'file2-fork',
             content=b'file2-line1-from-fork', message='commit2-fork',
             vcs_type=backend.alias, parent=commit1,
             newfile=True)
 
         commit_change(  # commit 3
             repo=r2_name, filename=b'file3-fork',
             content=b'file3-line1-from-fork', message='commit3-fork',
             vcs_type=backend.alias, parent=commit2, newfile=True)
 
         # compare !
         commit_id1 = repo1.scm_instance().DEFAULT_BRANCH_NAME
         commit_id2 = repo2.scm_instance().DEFAULT_BRANCH_NAME
 
         response = self.app.get(
             route_path('repo_compare',
                        repo_name=r2_name,
                        source_ref_type="branch", source_ref=commit_id1,
                        target_ref_type="branch", target_ref=commit_id2,
                        params=dict(merge='1', target_repo=r1_name),
                        ))
 
         response.mustcontain('%s@%s' % (r2_name, commit_id1))
         response.mustcontain('%s@%s' % (r1_name, commit_id2))
         response.mustcontain('No files')
         response.mustcontain('No commits in this compare')
 
         commit0 = commit_change(
             repo=r1_name, filename=b'file2',
             content=b'line1-added-after-fork', message='commit2-parent',
             vcs_type=backend.alias, parent=None, newfile=True)
 
         # compare !
         response = self.app.get(
             route_path('repo_compare',
                        repo_name=r2_name,
                        source_ref_type="branch", source_ref=commit_id1,
                        target_ref_type="branch", target_ref=commit_id2,
                        params=dict(merge='1', target_repo=r1_name),
                        ))
 
         response.mustcontain('%s@%s' % (r2_name, commit_id1))
         response.mustcontain('%s@%s' % (r1_name, commit_id2))
 
         response.mustcontain("""commit2-parent""")
         response.mustcontain("""line1-added-after-fork""")
         compare_page = ComparePage(response)
         compare_page.contains_change_summary(1, 1, 0)
 
     @pytest.mark.xfail_backends("svn")
     def test_compare_commits(self, backend, xhr_header):
         commit0 = backend.repo.get_commit(commit_idx=0)
         commit1 = backend.repo.get_commit(commit_idx=1)
 
         response = self.app.get(
             route_path('repo_compare',
                        repo_name=backend.repo_name,
                        source_ref_type="rev", source_ref=commit0.raw_id,
                        target_ref_type="rev", target_ref=commit1.raw_id,
                        params=dict(merge='1')
                        ),
             extra_environ=xhr_header, )
 
         # outgoing commits between those commits
         compare_page = ComparePage(response)
         compare_page.contains_commits(commits=[commit1])
 
     def test_errors_when_comparing_unknown_source_repo(self, backend):
         repo = backend.repo
 
         self.app.get(
             route_path('repo_compare',
                        repo_name='badrepo',
                        source_ref_type="rev", source_ref='tip',
                        target_ref_type="rev", target_ref='tip',
                        params=dict(merge='1', target_repo=repo.repo_name)
                        ),
             status=404)
 
     def test_errors_when_comparing_unknown_target_repo(self, backend):
         repo = backend.repo
         badrepo = 'badrepo'
 
         response = self.app.get(
             route_path('repo_compare',
                        repo_name=repo.repo_name,
                        source_ref_type="rev", source_ref='tip',
                        target_ref_type="rev", target_ref='tip',
                        params=dict(merge='1', target_repo=badrepo),
                        ),
             status=302)
         redirected = response.follow()
         redirected.mustcontain(
             'Could not find the target repo: `{}`'.format(badrepo))
 
     def test_compare_not_in_preview_mode(self, backend_stub):
         commit0 = backend_stub.repo.get_commit(commit_idx=0)
         commit1 = backend_stub.repo.get_commit(commit_idx=1)
 
         response = self.app.get(
             route_path('repo_compare',
                        repo_name=backend_stub.repo_name,
                        source_ref_type="rev", source_ref=commit0.raw_id,
                        target_ref_type="rev", target_ref=commit1.raw_id,
                        ))
 
         # outgoing commits between those commits
         compare_page = ComparePage(response)
         compare_page.swap_is_visible()
         compare_page.target_source_are_enabled()
 
     def test_compare_of_fork_with_largefiles(self, backend_hg, settings_util):
         orig = backend_hg.create_repo(number_of_commits=1)
         fork = backend_hg.create_fork()
 
         settings_util.create_repo_rhodecode_ui(
             orig, 'extensions', value='', key='largefiles', active=False)
         settings_util.create_repo_rhodecode_ui(
             fork, 'extensions', value='', key='largefiles', active=True)
 
         compare_module = ('rhodecode.lib.vcs.backends.hg.repository.'
                           'MercurialRepository.compare')
         with mock.patch(compare_module) as compare_mock:
             compare_mock.side_effect = RepositoryRequirementError()
 
             response = self.app.get(
                 route_path('repo_compare',
                            repo_name=orig.repo_name,
                            source_ref_type="rev", source_ref="tip",
                            target_ref_type="rev", target_ref="tip",
                            params=dict(merge='1', target_repo=fork.repo_name),
                            ),
                 status=302)
 
             assert_session_flash(
                 response,
                 'Could not compare repos with different large file settings')
 
 
 @pytest.mark.usefixtures("autologin_user")
 class TestCompareControllerSvn(object):
 
     def test_supports_references_with_path(self, app, backend_svn):
         repo = backend_svn['svn-simple-layout']
         commit_id = repo.get_commit(commit_idx=-1).raw_id
         response = app.get(
             route_path('repo_compare',
                        repo_name=repo.repo_name,
                        source_ref_type="tag",
                        source_ref="%s@%s" % ('tags/v0.1', commit_id),
                        target_ref_type="tag",
                        target_ref="%s@%s" % ('tags/v0.2', commit_id),
                        params=dict(merge='1'),
                        ),
             status=200)
 
         # Expecting no commits, since both paths are at the same revision
         response.mustcontain('No commits in this compare')
 
         # Should find only one file changed when comparing those two tags
         response.mustcontain('example.py')
         compare_page = ComparePage(response)
         compare_page.contains_change_summary(1, 5, 1)
 
     def test_shows_commits_if_different_ids(self, app, backend_svn):
         repo = backend_svn['svn-simple-layout']
         source_id = repo.get_commit(commit_idx=-6).raw_id
         target_id = repo.get_commit(commit_idx=-1).raw_id
         response = app.get(
             route_path('repo_compare',
                        repo_name=repo.repo_name,
                        source_ref_type="tag",
                        source_ref="%s@%s" % ('tags/v0.1', source_id),
                        target_ref_type="tag",
                        target_ref="%s@%s" % ('tags/v0.2', target_id),
                        params=dict(merge='1')
                        ),
             status=200)
 
         # It should show commits
         assert 'No commits in this compare' not in response.text
 
         # Should find only one file changed when comparing those two tags
         response.mustcontain('example.py')
         compare_page = ComparePage(response)
         compare_page.contains_change_summary(1, 5, 1)
 
 
 class ComparePage(AssertResponse):
     """
     Abstracts the page template from the tests
     """
 
     def contains_file_links_and_anchors(self, files):
         doc = lxml.html.fromstring(self.response.body)
         for filename, file_id in files:
             self.contains_one_anchor(file_id)
             diffblock = doc.cssselect('[data-f-path="%s"]' % filename)
             assert len(diffblock) == 2
             for lnk in diffblock[0].cssselect('a'):
                 if 'permalink' in lnk.text:
                     assert '#{}'.format(file_id) in lnk.attrib['href']
                     break
             else:
                 pytest.fail('Unable to find permalink')
 
     def contains_change_summary(self, files_changed, inserted, deleted):
         template = (
             '{files_changed} file{plural} changed: '
             '<span class="op-added">{inserted} inserted</span>, <span class="op-deleted">{deleted} deleted</span>')
         self.response.mustcontain(template.format(
             files_changed=files_changed,
             plural="s" if files_changed > 1 else "",
             inserted=inserted,
             deleted=deleted))
 
     def contains_commits(self, commits, ancestors=None):
         response = self.response
 
         for commit in commits:
             # Expecting to see the commit message in an element which
             # has the ID "c-{commit.raw_id}"
             self.element_contains('#c-' + commit.raw_id, commit.message)
             self.contains_one_link(
                 'r%s:%s' % (commit.idx, commit.short_id),
                 self._commit_url(commit))
 
         if ancestors:
             response.mustcontain('Ancestor')
             for ancestor in ancestors:
                 self.contains_one_link(
                     ancestor.short_id, self._commit_url(ancestor))
 
     def _commit_url(self, commit):
         return '/%s/changeset/%s' % (commit.repository.name, commit.raw_id)
 
     def swap_is_hidden(self):
         assert '<a id="btn-swap"' not in self.response.text
 
     def swap_is_visible(self):
         assert '<a id="btn-swap"' in self.response.text
 
     def target_source_are_disabled(self):
         response = self.response
         response.mustcontain("var enable_fields = false;")
         response.mustcontain('.select2("enable", enable_fields)')
 
     def target_source_are_enabled(self):
         response = self.response
         response.mustcontain("var enable_fields = true;")

rhodecode/apps/repository/tests/test_repo_pullrequests.py

@@ -1,1651 +1,1935 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 rhodecode
 from rhodecode.lib import helpers as h
-from rhodecode.lib.vcs.backends.base import MergeResponse, MergeFailureReason
+from rhodecode.lib.vcs.backends.base import MergeResponse, MergeFailureReason, Reference
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.lib.ext_json import json
 from rhodecode.model.changeset_status import ChangesetStatusModel
 from rhodecode.model.db import (
-    PullRequest, ChangesetStatus, UserLog, Notification, ChangesetComment, Repository)
+    PullRequest,
+    ChangesetStatus,
+    UserLog,
+    Notification,
+    ChangesetComment,
+    Repository,
+)
 from rhodecode.model.meta import Session
 from rhodecode.model.pull_request import PullRequestModel
 from rhodecode.model.user import UserModel
 from rhodecode.model.comment import CommentsModel
 from rhodecode.tests import (
-    assert_session_flash, TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN)
+    assert_session_flash,
+    TEST_USER_ADMIN_LOGIN,
+    TEST_USER_REGULAR_LOGIN,
+)
+from rhodecode.tests.fixture_mods.fixture_utils import PRTestUtility
 from rhodecode.tests.routes import route_path
 
 
[e-mail]('app', 'autologin_user')
[e-mail]("app", "autologin_user")
 @pytest.mark.backends("git", "hg")
 class TestPullrequestsView(object):
-
     def test_index(self, backend):
-        self.app.get(route_path(
-            'pullrequest_new',
-            repo_name=backend.repo_name))
+        self.app.get(route_path("pullrequest_new", repo_name=backend.repo_name))
 
     def test_option_menu_create_pull_request_exists(self, backend):
         repo_name = backend.repo_name
-        response = self.app.get(h.route_path('repo_summary', repo_name=repo_name))
+        response = self.app.get(h.route_path("repo_summary", repo_name=repo_name))
 
         create_pr_link = '<a href="%s">Create Pull Request</a>' % route_path(
-            'pullrequest_new', repo_name=repo_name)
+            "pullrequest_new", repo_name=repo_name
+        )
         response.mustcontain(create_pr_link)
 
     def test_create_pr_form_with_raw_commit_id(self, backend):
         repo = backend.repo
 
         self.app.get(
-            route_path('pullrequest_new', repo_name=repo.repo_name,
-                       commit=repo.get_commit().raw_id),
-            status=200)
+            route_path(
+                "pullrequest_new",
+                repo_name=repo.repo_name,
+                commit=repo.get_commit().raw_id,
+            ),
+            status=200,
+        )
 
-    @pytest.mark.parametrize('pr_merge_enabled', [True, False])
-    @pytest.mark.parametrize('range_diff', ["0", "1"])
+    @pytest.mark.parametrize("pr_merge_enabled", [True, False])
+    @pytest.mark.parametrize("range_diff", ["0", "1"])
     def test_show(self, pr_util, pr_merge_enabled, range_diff):
         pull_request = pr_util.create_pull_request(
-            mergeable=pr_merge_enabled, enable_notifications=False)
+            mergeable=pr_merge_enabled, enable_notifications=False
+        )
 
-        response = self.app.get(route_path(
-            'pullrequest_show',
-            repo_name=pull_request.target_repo.scm_instance().name,
-            pull_request_id=pull_request.pull_request_id,
-            params={'range-diff': range_diff}))
+        response = self.app.get(
+            route_path(
+                "pullrequest_show",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request.pull_request_id,
+                params={"range-diff": range_diff},
+            )
+        )
 
         for commit_id in pull_request.revisions:
             response.mustcontain(commit_id)
 
         response.mustcontain(pull_request.target_ref_parts.type)
         response.mustcontain(pull_request.target_ref_parts.name)
 
         response.mustcontain('class="pull-request-merge"')
 
         if pr_merge_enabled:
-            response.mustcontain('Pull request reviewer approval is pending')
+            response.mustcontain("Pull request reviewer approval is pending")
         else:
-            response.mustcontain('Server-side pull request merging is disabled.')
+            response.mustcontain("Server-side pull request merging is disabled.")
 
         if range_diff == "1":
-            response.mustcontain('Turn off: Show the diff as commit range')
+            response.mustcontain("Turn off: Show the diff as commit range")
 
     def test_show_versions_of_pr(self, backend, csrf_token):
         commits = [
-            {'message': 'initial-commit',
-             'added': [FileNode(b'test-file.txt', b'LINE1\n')]},
-
-            {'message': 'commit-1',
-             'changed': [FileNode(b'test-file.txt', b'LINE1\nLINE2\n')]},
+            {
+                "message": "initial-commit",
+                "added": [FileNode(b"test-file.txt", b"LINE1\n")],
+            },
+            {
+                "message": "commit-1",
+                "changed": [FileNode(b"test-file.txt", b"LINE1\nLINE2\n")],
+            },
             # Above is the initial version of PR that changes a single line
-
             # from now on we'll add 3x commit adding a nother line on each step
-            {'message': 'commit-2',
-             'changed': [FileNode(b'test-file.txt', b'LINE1\nLINE2\nLINE3\n')]},
-
-            {'message': 'commit-3',
-             'changed': [FileNode(b'test-file.txt', b'LINE1\nLINE2\nLINE3\nLINE4\n')]},
-
-            {'message': 'commit-4',
-             'changed': [FileNode(b'test-file.txt', b'LINE1\nLINE2\nLINE3\nLINE4\nLINE5\n')]},
+            {
+                "message": "commit-2",
+                "changed": [FileNode(b"test-file.txt", b"LINE1\nLINE2\nLINE3\n")],
+            },
+            {
+                "message": "commit-3",
+                "changed": [
+                    FileNode(b"test-file.txt", b"LINE1\nLINE2\nLINE3\nLINE4\n")
+                ],
+            },
+            {
+                "message": "commit-4",
+                "changed": [
+                    FileNode(b"test-file.txt", b"LINE1\nLINE2\nLINE3\nLINE4\nLINE5\n")
+                ],
+            },
         ]
 
         commit_ids = backend.create_master_repo(commits)
-        target = backend.create_repo(heads=['initial-commit'])
-        source = backend.create_repo(heads=['commit-1'])
+        target = backend.create_repo(heads=["initial-commit"])
+        source = backend.create_repo(heads=["commit-1"])
         source_repo_name = source.repo_name
         target_repo_name = target.repo_name
 
-        target_ref = 'branch:{branch}:{commit_id}'.format(
-            branch=backend.default_branch_name, commit_id=commit_ids['initial-commit'])
-        source_ref = 'branch:{branch}:{commit_id}'.format(
-             branch=backend.default_branch_name, commit_id=commit_ids['commit-1'])
+        target_ref = "branch:{branch}:{commit_id}".format(
+            branch=backend.default_branch_name, commit_id=commit_ids["initial-commit"]
+        )
+        source_ref = "branch:{branch}:{commit_id}".format(
+            branch=backend.default_branch_name, commit_id=commit_ids["commit-1"]
+        )
 
         response = self.app.post(
-            route_path('pullrequest_create', repo_name=source.repo_name),
+            route_path("pullrequest_create", repo_name=source.repo_name),
             [
-                ('source_repo', source_repo_name),
-                ('source_ref', source_ref),
-                ('target_repo', target_repo_name),
-                ('target_ref',  target_ref),
-                ('common_ancestor', commit_ids['initial-commit']),
-                ('pullrequest_title', 'Title'),
-                ('pullrequest_desc', 'Description'),
-                ('description_renderer', 'markdown'),
-                ('__start__', 'review_members:sequence'),
-                    ('__start__', 'reviewer:mapping'),
-                        ('user_id', '1'),
-                        ('__start__', 'reasons:sequence'),
-                            ('reason', 'Some reason'),
-                        ('__end__', 'reasons:sequence'),
-                        ('__start__', 'rules:sequence'),
-                        ('__end__', 'rules:sequence'),
-                        ('mandatory', 'False'),
-                    ('__end__', 'reviewer:mapping'),
-                ('__end__', 'review_members:sequence'),
-                ('__start__', 'revisions:sequence'),
-                    ('revisions', commit_ids['commit-1']),
-                ('__end__', 'revisions:sequence'),
-                ('user', ''),
-                ('csrf_token', csrf_token),
+                ("source_repo", source_repo_name),
+                ("source_ref", source_ref),
+                ("target_repo", target_repo_name),
+                ("target_ref", target_ref),
+                ("common_ancestor", commit_ids["initial-commit"]),
+                ("pullrequest_title", "Title"),
+                ("pullrequest_desc", "Description"),
+                ("description_renderer", "markdown"),
+                ("__start__", "review_members:sequence"),
+                ("__start__", "reviewer:mapping"),
+                ("user_id", "1"),
+                ("__start__", "reasons:sequence"),
+                ("reason", "Some reason"),
+                ("__end__", "reasons:sequence"),
+                ("__start__", "rules:sequence"),
+                ("__end__", "rules:sequence"),
+                ("mandatory", "False"),
+                ("__end__", "reviewer:mapping"),
+                ("__end__", "review_members:sequence"),
+                ("__start__", "revisions:sequence"),
+                ("revisions", commit_ids["commit-1"]),
+                ("__end__", "revisions:sequence"),
+                ("user", ""),
+                ("csrf_token", csrf_token),
             ],
-            status=302)
+            status=302,
+        )
 
-        location = response.headers['Location']
+        location = response.headers["Location"]
 
-        pull_request_id = location.rsplit('/', 1)[1]
-        assert pull_request_id != 'new'
+        pull_request_id = location.rsplit("/", 1)[1]
+        assert pull_request_id != "new"
         pull_request = PullRequest.get(int(pull_request_id))
 
         pull_request_id = pull_request.pull_request_id
 
         # Show initial version of PR
         response = self.app.get(
-            route_path('pullrequest_show',
-                       repo_name=target_repo_name,
-                       pull_request_id=pull_request_id))
+            route_path(
+                "pullrequest_show",
+                repo_name=target_repo_name,
+                pull_request_id=pull_request_id,
+            )
+        )
 
-        response.mustcontain('commit-1')
-        response.mustcontain(no=['commit-2'])
-        response.mustcontain(no=['commit-3'])
-        response.mustcontain(no=['commit-4'])
+        response.mustcontain("commit-1")
+        response.mustcontain(no=["commit-2"])
+        response.mustcontain(no=["commit-3"])
+        response.mustcontain(no=["commit-4"])
 
         response.mustcontain('cb-addition"></span><span>LINE2</span>')
-        response.mustcontain(no=['LINE3'])
-        response.mustcontain(no=['LINE4'])
-        response.mustcontain(no=['LINE5'])
+        response.mustcontain(no=["LINE3"])
+        response.mustcontain(no=["LINE4"])
+        response.mustcontain(no=["LINE5"])
 
         # update PR #1
         source_repo = Repository.get_by_repo_name(source_repo_name)
-        backend.pull_heads(source_repo, heads=['commit-2'])
+        backend.pull_heads(source_repo, heads=["commit-2"])
         response = self.app.post(
-            route_path('pullrequest_update',
-                       repo_name=target_repo_name, pull_request_id=pull_request_id),
-            params={'update_commits': 'true', 'csrf_token': csrf_token})
+            route_path(
+                "pullrequest_update",
+                repo_name=target_repo_name,
+                pull_request_id=pull_request_id,
+            ),
+            params={"update_commits": "true", "csrf_token": csrf_token},
+        )
 
         # update PR #2
         source_repo = Repository.get_by_repo_name(source_repo_name)
-        backend.pull_heads(source_repo, heads=['commit-3'])
+        backend.pull_heads(source_repo, heads=["commit-3"])
         response = self.app.post(
-            route_path('pullrequest_update',
-                       repo_name=target_repo_name, pull_request_id=pull_request_id),
-            params={'update_commits': 'true', 'csrf_token': csrf_token})
+            route_path(
+                "pullrequest_update",
+                repo_name=target_repo_name,
+                pull_request_id=pull_request_id,
+            ),
+            params={"update_commits": "true", "csrf_token": csrf_token},
+        )
 
         # update PR #3
         source_repo = Repository.get_by_repo_name(source_repo_name)
-        backend.pull_heads(source_repo, heads=['commit-4'])
+        backend.pull_heads(source_repo, heads=["commit-4"])
         response = self.app.post(
-            route_path('pullrequest_update',
-                       repo_name=target_repo_name, pull_request_id=pull_request_id),
-            params={'update_commits': 'true', 'csrf_token': csrf_token})
+            route_path(
+                "pullrequest_update",
+                repo_name=target_repo_name,
+                pull_request_id=pull_request_id,
+            ),
+            params={"update_commits": "true", "csrf_token": csrf_token},
+        )
 
         # Show final version !
         response = self.app.get(
-            route_path('pullrequest_show',
-                       repo_name=target_repo_name,
-                       pull_request_id=pull_request_id))
+            route_path(
+                "pullrequest_show",
+                repo_name=target_repo_name,
+                pull_request_id=pull_request_id,
+            )
+        )
 
         # 3 updates, and the latest == 4
-        response.mustcontain('4 versions available for this pull request')
-        response.mustcontain(no=['rhodecode diff rendering error'])
+        response.mustcontain("4 versions available for this pull request")
+        response.mustcontain(no=["rhodecode diff rendering error"])
 
         # initial show must have 3 commits, and 3 adds
-        response.mustcontain('commit-1')
-        response.mustcontain('commit-2')
-        response.mustcontain('commit-3')
-        response.mustcontain('commit-4')
+        response.mustcontain("commit-1")
+        response.mustcontain("commit-2")
+        response.mustcontain("commit-3")
+        response.mustcontain("commit-4")
 
         response.mustcontain('cb-addition"></span><span>LINE2</span>')
         response.mustcontain('cb-addition"></span><span>LINE3</span>')
         response.mustcontain('cb-addition"></span><span>LINE4</span>')
         response.mustcontain('cb-addition"></span><span>LINE5</span>')
 
         # fetch versions
         pr = PullRequest.get(pull_request_id)
         versions = [x.pull_request_version_id for x in pr.versions.all()]
         assert len(versions) == 3
 
         # show v1,v2,v3,v4
         def cb_line(text):
             return 'cb-addition"></span><span>{}</span>'.format(text)
 
         def cb_context(text):
-            return '<span class="cb-code"><span class="cb-action cb-context">' \
-                   '</span><span>{}</span></span>'.format(text)
+            return (
+                '<span class="cb-code"><span class="cb-action cb-context">'
+                "</span><span>{}</span></span>".format(text)
+            )
 
         commit_tests = {
             # in response, not in response
-            1: (['commit-1'], ['commit-2', 'commit-3', 'commit-4']),
-            2: (['commit-1', 'commit-2'], ['commit-3', 'commit-4']),
-            3: (['commit-1', 'commit-2', 'commit-3'], ['commit-4']),
-            4: (['commit-1', 'commit-2', 'commit-3', 'commit-4'], []),
+            1: (["commit-1"], ["commit-2", "commit-3", "commit-4"]),
+            2: (["commit-1", "commit-2"], ["commit-3", "commit-4"]),
+            3: (["commit-1", "commit-2", "commit-3"], ["commit-4"]),
+            4: (["commit-1", "commit-2", "commit-3", "commit-4"], []),
         }
         diff_tests = {
-            1: (['LINE2'], ['LINE3', 'LINE4', 'LINE5']),
-            2: (['LINE2', 'LINE3'], ['LINE4', 'LINE5']),
-            3: (['LINE2', 'LINE3', 'LINE4'], ['LINE5']),
-            4: (['LINE2', 'LINE3', 'LINE4', 'LINE5'], []),
+            1: (["LINE2"], ["LINE3", "LINE4", "LINE5"]),
+            2: (["LINE2", "LINE3"], ["LINE4", "LINE5"]),
+            3: (["LINE2", "LINE3", "LINE4"], ["LINE5"]),
+            4: (["LINE2", "LINE3", "LINE4", "LINE5"], []),
         }
         for idx, ver in enumerate(versions, 1):
+            response = self.app.get(
+                route_path(
+                    "pullrequest_show",
+                    repo_name=target_repo_name,
+                    pull_request_id=pull_request_id,
+                    params={"version": ver},
+                )
+            )
 
-            response = self.app.get(
-                route_path('pullrequest_show',
-                           repo_name=target_repo_name,
-                           pull_request_id=pull_request_id,
-                           params={'version': ver}))
-
-            response.mustcontain(no=['rhodecode diff rendering error'])
-            response.mustcontain('Showing changes at v{}'.format(idx))
+            response.mustcontain(no=["rhodecode diff rendering error"])
+            response.mustcontain("Showing changes at v{}".format(idx))
 
             yes, no = commit_tests[idx]
             for y in yes:
                 response.mustcontain(y)
             for n in no:
                 response.mustcontain(no=n)
 
             yes, no = diff_tests[idx]
             for y in yes:
                 response.mustcontain(cb_line(y))
             for n in no:
                 response.mustcontain(no=n)
 
         # show diff between versions
         diff_compare_tests = {
-            1: (['LINE3'], ['LINE1', 'LINE2']),
-            2: (['LINE3', 'LINE4'], ['LINE1', 'LINE2']),
-            3: (['LINE3', 'LINE4', 'LINE5'], ['LINE1', 'LINE2']),
+            1: (["LINE3"], ["LINE1", "LINE2"]),
+            2: (["LINE3", "LINE4"], ["LINE1", "LINE2"]),
+            3: (["LINE3", "LINE4", "LINE5"], ["LINE1", "LINE2"]),
         }
         for idx, ver in enumerate(versions, 1):
             adds, context = diff_compare_tests[idx]
 
-            to_ver = ver+1
+            to_ver = ver + 1
             if idx == 3:
-                to_ver = 'latest'
+                to_ver = "latest"
 
             response = self.app.get(
-                route_path('pullrequest_show',
-                           repo_name=target_repo_name,
-                           pull_request_id=pull_request_id,
-                           params={'from_version': versions[0], 'version': to_ver}))
+                route_path(
+                    "pullrequest_show",
+                    repo_name=target_repo_name,
+                    pull_request_id=pull_request_id,
+                    params={"from_version": versions[0], "version": to_ver},
+                )
+            )
 
-            response.mustcontain(no=['rhodecode diff rendering error'])
+            response.mustcontain(no=["rhodecode diff rendering error"])
 
             for a in adds:
                 response.mustcontain(cb_line(a))
             for c in context:
                 response.mustcontain(cb_context(c))
 
         # test version v2 -> v3
         response = self.app.get(
-            route_path('pullrequest_show',
-                       repo_name=target_repo_name,
-                       pull_request_id=pull_request_id,
-                       params={'from_version': versions[1], 'version': versions[2]}))
+            route_path(
+                "pullrequest_show",
+                repo_name=target_repo_name,
+                pull_request_id=pull_request_id,
+                params={"from_version": versions[1], "version": versions[2]},
+            )
+        )
 
-        response.mustcontain(cb_context('LINE1'))
-        response.mustcontain(cb_context('LINE2'))
-        response.mustcontain(cb_context('LINE3'))
-        response.mustcontain(cb_line('LINE4'))
+        response.mustcontain(cb_context("LINE1"))
+        response.mustcontain(cb_context("LINE2"))
+        response.mustcontain(cb_context("LINE3"))
+        response.mustcontain(cb_line("LINE4"))
 
     def test_close_status_visibility(self, pr_util, user_util, csrf_token):
         # Logout
         response = self.app.post(
-            h.route_path('logout'),
-            params={'csrf_token': csrf_token})
+            h.route_path("logout"), params={"csrf_token": csrf_token}
+        )
         # Login as regular user
-        response = self.app.post(h.route_path('login'),
-                                 {'username': TEST_USER_REGULAR_LOGIN,
-                                  'password': 'test12'})
+        response = self.app.post(
+            h.route_path("login"),
+            {"username": TEST_USER_REGULAR_LOGIN, "password": "test12"},
+        )
+
+        pull_request = pr_util.create_pull_request(author=TEST_USER_REGULAR_LOGIN)
 
-        pull_request = pr_util.create_pull_request(
-            author=TEST_USER_REGULAR_LOGIN)
+        response = self.app.get(
+            route_path(
+                "pullrequest_show",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request.pull_request_id,
+            )
+        )
 
-        response = self.app.get(route_path(
-            'pullrequest_show',
-            repo_name=pull_request.target_repo.scm_instance().name,
-            pull_request_id=pull_request.pull_request_id))
-
-        response.mustcontain('Server-side pull request merging is disabled.')
+        response.mustcontain("Server-side pull request merging is disabled.")
 
         assert_response = response.assert_response()
         # for regular user without a merge permissions, we don't see it
-        assert_response.no_element_exists('#close-pull-request-action')
+        assert_response.no_element_exists("#close-pull-request-action")
 
         user_util.grant_user_permission_to_repo(
             pull_request.target_repo,
             UserModel().get_by_username(TEST_USER_REGULAR_LOGIN),
-            'repository.write')
-        response = self.app.get(route_path(
-            'pullrequest_show',
-            repo_name=pull_request.target_repo.scm_instance().name,
-            pull_request_id=pull_request.pull_request_id))
+            "repository.write",
+        )
+        response = self.app.get(
+            route_path(
+                "pullrequest_show",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request.pull_request_id,
+            )
+        )
 
-        response.mustcontain('Server-side pull request merging is disabled.')
+        response.mustcontain("Server-side pull request merging is disabled.")
 
         assert_response = response.assert_response()
         # now regular user has a merge permissions, we have CLOSE button
-        assert_response.one_element_exists('#close-pull-request-action')
+        assert_response.one_element_exists("#close-pull-request-action")
 
     def test_show_invalid_commit_id(self, pr_util):
         # Simulating invalid revisions which will cause a lookup error
         pull_request = pr_util.create_pull_request()
-        pull_request.revisions = ['invalid']
+        pull_request.revisions = ["invalid"]
         Session().add(pull_request)
         Session().commit()
 
-        response = self.app.get(route_path(
-            'pullrequest_show',
-            repo_name=pull_request.target_repo.scm_instance().name,
-            pull_request_id=pull_request.pull_request_id))
+        response = self.app.get(
+            route_path(
+                "pullrequest_show",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request.pull_request_id,
+            )
+        )
 
         for commit_id in pull_request.revisions:
             response.mustcontain(commit_id)
 
     def test_show_invalid_source_reference(self, pr_util):
         pull_request = pr_util.create_pull_request()
-        pull_request.source_ref = 'branch:b:invalid'
+        pull_request.source_ref = "branch:b:invalid"
         Session().add(pull_request)
         Session().commit()
 
-        self.app.get(route_path(
-            'pullrequest_show',
-            repo_name=pull_request.target_repo.scm_instance().name,
-            pull_request_id=pull_request.pull_request_id))
+        self.app.get(
+            route_path(
+                "pullrequest_show",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request.pull_request_id,
+            )
+        )
 
     def test_edit_title_description(self, pr_util, csrf_token):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
 
         response = self.app.post(
-            route_path('pullrequest_update',
-                       repo_name=pull_request.target_repo.repo_name,
-                       pull_request_id=pull_request_id),
+            route_path(
+                "pullrequest_update",
+                repo_name=pull_request.target_repo.repo_name,
+                pull_request_id=pull_request_id,
+            ),
             params={
-                'edit_pull_request': 'true',
-                'title': 'New title',
-                'description': 'New description',
-                'csrf_token': csrf_token})
+                "edit_pull_request": "true",
+                "title": "New title",
+                "description": "New description",
+                "csrf_token": csrf_token,
+            },
+        )
 
         assert_session_flash(
-            response, 'Pull request title & description updated.',
-            category='success')
+            response, "Pull request title & description updated.", category="success"
+        )
 
         pull_request = PullRequest.get(pull_request_id)
-        assert pull_request.title == 'New title'
-        assert pull_request.description == 'New description'
+        assert pull_request.title == "New title"
+        assert pull_request.description == "New description"
 
     def test_edit_title_description_special(self, pr_util, csrf_token):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
 
         response = self.app.post(
-            route_path('pullrequest_update',
-                       repo_name=pull_request.target_repo.repo_name,
-                       pull_request_id=pull_request_id),
+            route_path(
+                "pullrequest_update",
+                repo_name=pull_request.target_repo.repo_name,
+                pull_request_id=pull_request_id,
+            ),
             params={
-                'edit_pull_request': 'true',
-                'title': 'New title {} {2} {foo}',
-                'description': 'New description',
-                'csrf_token': csrf_token})
+                "edit_pull_request": "true",
+                "title": "New title {} {2} {foo}",
+                "description": "New description",
+                "csrf_token": csrf_token,
+            },
+        )
 
         assert_session_flash(
-            response, 'Pull request title & description updated.',
-            category='success')
+            response, "Pull request title & description updated.", category="success"
+        )
 
         pull_request = PullRequest.get(pull_request_id)
-        assert pull_request.title_safe == 'New title {{}} {{2}} {{foo}}'
+        assert pull_request.title_safe == "New title {{}} {{2}} {{foo}}"
 
     def test_edit_title_description_closed(self, pr_util, csrf_token):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
         repo_name = pull_request.target_repo.repo_name
         pr_util.close()
 
         response = self.app.post(
-            route_path('pullrequest_update',
-                       repo_name=repo_name, pull_request_id=pull_request_id),
+            route_path(
+                "pullrequest_update",
+                repo_name=repo_name,
+                pull_request_id=pull_request_id,
+            ),
             params={
-                'edit_pull_request': 'true',
-                'title': 'New title',
-                'description': 'New description',
-                'csrf_token': csrf_token}, status=200)
+                "edit_pull_request": "true",
+                "title": "New title",
+                "description": "New description",
+                "csrf_token": csrf_token,
+            },
+            status=200,
+        )
         assert_session_flash(
-            response, 'Cannot update closed pull requests.',
-            category='error')
+            response, "Cannot update closed pull requests.", category="error"
+        )
 
     def test_update_invalid_source_reference(self, pr_util, csrf_token):
         from rhodecode.lib.vcs.backends.base import UpdateFailureReason
 
         pull_request = pr_util.create_pull_request()
-        pull_request.source_ref = 'branch:invalid-branch:invalid-commit-id'
+        pull_request.source_ref = "branch:invalid-branch:invalid-commit-id"
         Session().add(pull_request)
         Session().commit()
 
         pull_request_id = pull_request.pull_request_id
 
         response = self.app.post(
-            route_path('pullrequest_update',
+            route_path(
+                "pullrequest_update",
                 repo_name=pull_request.target_repo.repo_name,
-                pull_request_id=pull_request_id),
-            params={'update_commits': 'true', 'csrf_token': csrf_token})
+                pull_request_id=pull_request_id,
+            ),
+            params={"update_commits": "true", "csrf_token": csrf_token},
+        )
 
-        expected_msg = str(PullRequestModel.UPDATE_STATUS_MESSAGES[
-            UpdateFailureReason.MISSING_SOURCE_REF])
-        assert_session_flash(response, expected_msg, category='error')
+        expected_msg = str(
+            PullRequestModel.UPDATE_STATUS_MESSAGES[
+                UpdateFailureReason.MISSING_SOURCE_REF
+            ]
+        )
+        assert_session_flash(response, expected_msg, category="error")
 
     def test_missing_target_reference(self, pr_util, csrf_token):
         from rhodecode.lib.vcs.backends.base import MergeFailureReason
-        pull_request = pr_util.create_pull_request(
-            approved=True, mergeable=True)
-        unicode_reference = 'branch:invalid-branch:invalid-commit-id'
+
+        pull_request = pr_util.create_pull_request(approved=True, mergeable=True)
+        unicode_reference = "branch:invalid-branch:invalid-commit-id"
         pull_request.target_ref = unicode_reference
         Session().add(pull_request)
         Session().commit()
 
         pull_request_id = pull_request.pull_request_id
         pull_request_url = route_path(
-            'pullrequest_show',
+            "pullrequest_show",
             repo_name=pull_request.target_repo.repo_name,
-            pull_request_id=pull_request_id)
+            pull_request_id=pull_request_id,
+        )
 
         response = self.app.get(pull_request_url)
-        target_ref_id = 'invalid-branch'
+        # target_ref_id = "invalid-branch"
+
         merge_resp = MergeResponse(
-            True, True, '', MergeFailureReason.MISSING_TARGET_REF,
-            metadata={'target_ref': PullRequest.unicode_to_reference(unicode_reference)})
+            True,
+            True,
+            Reference("commit", "STUB_COMMIT_ID", "STUB_COMMIT_ID"),
+            MergeFailureReason.MISSING_TARGET_REF,
+            metadata={
+                "target_ref": PullRequest.unicode_to_reference(unicode_reference)
+            },
+        )
         response.assert_response().element_contains(
-            'div[data-role="merge-message"]', merge_resp.merge_status_message)
+            'div[data-role="merge-message"]', merge_resp.merge_status_message
+        )
 
     def test_comment_and_close_pull_request_custom_message_approved(
-            self, pr_util, csrf_token, xhr_header):
-
+        self, pr_util, csrf_token, xhr_header
+    ):
         pull_request = pr_util.create_pull_request(approved=True)
         pull_request_id = pull_request.pull_request_id
         author = pull_request.user_id
         repo = pull_request.target_repo.repo_id
 
         self.app.post(
-            route_path('pullrequest_comment_create',
-                       repo_name=pull_request.target_repo.scm_instance().name,
-                       pull_request_id=pull_request_id),
+            route_path(
+                "pullrequest_comment_create",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request_id,
+            ),
             params={
-                'close_pull_request': '1',
-                'text': 'Closing a PR',
-                'csrf_token': csrf_token},
-            extra_environ=xhr_header,)
+                "close_pull_request": "1",
+                "text": "Closing a PR",
+                "csrf_token": csrf_token,
+            },
+            extra_environ=xhr_header,
+        )
 
-        journal = UserLog.query()\
-            .filter(UserLog.user_id == author)\
-            .filter(UserLog.repository_id == repo) \
-            .order_by(UserLog.user_log_id.asc()) \
+        journal = (
+            UserLog.query()
+            .filter(UserLog.user_id == author)
+            .filter(UserLog.repository_id == repo)
+            .order_by(UserLog.user_log_id.asc())
             .all()
-        assert journal[-1].action == 'repo.pull_request.close'
+        )
+        assert journal[-1].action == "repo.pull_request.close"
 
         pull_request = PullRequest.get(pull_request_id)
         assert pull_request.is_closed()
 
         status = ChangesetStatusModel().get_status(
-            pull_request.source_repo, pull_request=pull_request)
+            pull_request.source_repo, pull_request=pull_request
+        )
         assert status == ChangesetStatus.STATUS_APPROVED
-        comments = ChangesetComment().query() \
-            .filter(ChangesetComment.pull_request == pull_request) \
-            .order_by(ChangesetComment.comment_id.asc())\
+        comments = (
+            ChangesetComment()
+            .query()
+            .filter(ChangesetComment.pull_request == pull_request)
+            .order_by(ChangesetComment.comment_id.asc())
             .all()
-        assert comments[-1].text == 'Closing a PR'
+        )
+        assert comments[-1].text == "Closing a PR"
 
     def test_comment_force_close_pull_request_rejected(
-            self, pr_util, csrf_token, xhr_header):
+        self, pr_util, csrf_token, xhr_header
+    ):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
         PullRequestModel().update_reviewers(
-            pull_request_id, [
-                (1, ['reason'], False, 'reviewer', []),
-                (2, ['reason2'], False, 'reviewer', [])],
-            pull_request.author)
+            pull_request_id,
+            [
+                (1, ["reason"], False, "reviewer", []),
+                (2, ["reason2"], False, "reviewer", []),
+            ],
+            pull_request.author,
+        )
         author = pull_request.user_id
         repo = pull_request.target_repo.repo_id
 
         self.app.post(
-            route_path('pullrequest_comment_create',
+            route_path(
+                "pullrequest_comment_create",
                 repo_name=pull_request.target_repo.scm_instance().name,
-                pull_request_id=pull_request_id),
-            params={
-                'close_pull_request': '1',
-                'csrf_token': csrf_token},
-            extra_environ=xhr_header)
+                pull_request_id=pull_request_id,
+            ),
+            params={"close_pull_request": "1", "csrf_token": csrf_token},
+            extra_environ=xhr_header,
+        )
 
         pull_request = PullRequest.get(pull_request_id)
 
-        journal = UserLog.query()\
-            .filter(UserLog.user_id == author, UserLog.repository_id == repo) \
-            .order_by(UserLog.user_log_id.asc()) \
+        journal = (
+            UserLog.query()
+            .filter(UserLog.user_id == author, UserLog.repository_id == repo)
+            .order_by(UserLog.user_log_id.asc())
             .all()
-        assert journal[-1].action == 'repo.pull_request.close'
+        )
+        assert journal[-1].action == "repo.pull_request.close"
 
         # check only the latest status, not the review status
         status = ChangesetStatusModel().get_status(
-            pull_request.source_repo, pull_request=pull_request)
+            pull_request.source_repo, pull_request=pull_request
+        )
         assert status == ChangesetStatus.STATUS_REJECTED
 
-    def test_comment_and_close_pull_request(
-            self, pr_util, csrf_token, xhr_header):
+    def test_comment_and_close_pull_request(self, pr_util, csrf_token, xhr_header):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
 
         response = self.app.post(
-            route_path('pullrequest_comment_create',
-                       repo_name=pull_request.target_repo.scm_instance().name,
-                       pull_request_id=pull_request.pull_request_id),
-            params={
-                'close_pull_request': 'true',
-                'csrf_token': csrf_token},
-            extra_environ=xhr_header)
+            route_path(
+                "pullrequest_comment_create",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request.pull_request_id,
+            ),
+            params={"close_pull_request": "true", "csrf_token": csrf_token},
+            extra_environ=xhr_header,
+        )
 
         assert response.json
 
         pull_request = PullRequest.get(pull_request_id)
         assert pull_request.is_closed()
 
         # check only the latest status, not the review status
         status = ChangesetStatusModel().get_status(
-            pull_request.source_repo, pull_request=pull_request)
+            pull_request.source_repo, pull_request=pull_request
+        )
         assert status == ChangesetStatus.STATUS_REJECTED
 
     def test_comment_and_close_pull_request_try_edit_comment(
-            self, pr_util, csrf_token, xhr_header
+        self, pr_util, csrf_token, xhr_header
     ):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
         target_scm = pull_request.target_repo.scm_instance()
         target_scm_name = target_scm.name
 
         response = self.app.post(
             route_path(
-                'pullrequest_comment_create',
+                "pullrequest_comment_create",
                 repo_name=target_scm_name,
                 pull_request_id=pull_request_id,
             ),
             params={
-                'close_pull_request': 'true',
-                'csrf_token': csrf_token,
+                "close_pull_request": "true",
+                "csrf_token": csrf_token,
             },
-            extra_environ=xhr_header)
+            extra_environ=xhr_header,
+        )
 
         assert response.json
 
         pull_request = PullRequest.get(pull_request_id)
         target_scm = pull_request.target_repo.scm_instance()
         target_scm_name = target_scm.name
         assert pull_request.is_closed()
 
         # check only the latest status, not the review status
         status = ChangesetStatusModel().get_status(
-            pull_request.source_repo, pull_request=pull_request)
+            pull_request.source_repo, pull_request=pull_request
+        )
         assert status == ChangesetStatus.STATUS_REJECTED
 
         for comment_id in response.json.keys():
-            test_text = 'test'
+            test_text = "test"
             response = self.app.post(
                 route_path(
-                    'pullrequest_comment_edit',
+                    "pullrequest_comment_edit",
                     repo_name=target_scm_name,
                     pull_request_id=pull_request_id,
                     comment_id=comment_id,
                 ),
                 extra_environ=xhr_header,
                 params={
-                    'csrf_token': csrf_token,
-                    'text': test_text,
+                    "csrf_token": csrf_token,
+                    "text": test_text,
                 },
                 status=403,
             )
             assert response.status_int == 403
 
     def test_comment_and_comment_edit(self, pr_util, csrf_token, xhr_header):
         pull_request = pr_util.create_pull_request()
         target_scm = pull_request.target_repo.scm_instance()
         target_scm_name = target_scm.name
 
         response = self.app.post(
             route_path(
-                'pullrequest_comment_create',
+                "pullrequest_comment_create",
                 repo_name=target_scm_name,
-                pull_request_id=pull_request.pull_request_id),
+                pull_request_id=pull_request.pull_request_id,
+            ),
             params={
-                'csrf_token': csrf_token,
-                'text': 'init',
+                "csrf_token": csrf_token,
+                "text": "init",
             },
             extra_environ=xhr_header,
         )
         assert response.json
 
         for comment_id in response.json.keys():
             assert comment_id
-            test_text = 'test'
+            test_text = "test"
             self.app.post(
                 route_path(
-                    'pullrequest_comment_edit',
+                    "pullrequest_comment_edit",
                     repo_name=target_scm_name,
                     pull_request_id=pull_request.pull_request_id,
                     comment_id=comment_id,
                 ),
                 extra_environ=xhr_header,
                 params={
-                    'csrf_token': csrf_token,
-                    'text': test_text,
-                    'version': '0',
+                    "csrf_token": csrf_token,
+                    "text": test_text,
+                    "version": "0",
                 },
-
             )
-            text_form_db = ChangesetComment.query().filter(
-                ChangesetComment.comment_id == comment_id).first().text
+            text_form_db = (
+                ChangesetComment.query()
+                .filter(ChangesetComment.comment_id == comment_id)
+                .first()
+                .text
+            )
             assert test_text == text_form_db
 
     def test_comment_and_comment_edit_special(self, pr_util, csrf_token, xhr_header):
         pull_request = pr_util.create_pull_request()
         target_scm = pull_request.target_repo.scm_instance()
         target_scm_name = target_scm.name
 
         response = self.app.post(
             route_path(
-                'pullrequest_comment_create',
+                "pullrequest_comment_create",
                 repo_name=target_scm_name,
-                pull_request_id=pull_request.pull_request_id),
+                pull_request_id=pull_request.pull_request_id,
+            ),
             params={
-                'csrf_token': csrf_token,
-                'text': 'init',
+                "csrf_token": csrf_token,
+                "text": "init",
             },
             extra_environ=xhr_header,
         )
         assert response.json
 
         for comment_id in response.json.keys():
-            test_text = 'init'
+            test_text = "init"
             response = self.app.post(
                 route_path(
-                    'pullrequest_comment_edit',
+                    "pullrequest_comment_edit",
                     repo_name=target_scm_name,
                     pull_request_id=pull_request.pull_request_id,
                     comment_id=comment_id,
                 ),
                 extra_environ=xhr_header,
                 params={
-                    'csrf_token': csrf_token,
-                    'text': test_text,
-                    'version': '0',
+                    "csrf_token": csrf_token,
+                    "text": test_text,
+                    "version": "0",
                 },
                 status=404,
-
             )
             assert response.status_int == 404
 
     def test_comment_and_try_edit_already_edited(self, pr_util, csrf_token, xhr_header):
         pull_request = pr_util.create_pull_request()
         target_scm = pull_request.target_repo.scm_instance()
         target_scm_name = target_scm.name
 
         response = self.app.post(
             route_path(
-                'pullrequest_comment_create',
+                "pullrequest_comment_create",
                 repo_name=target_scm_name,
-                pull_request_id=pull_request.pull_request_id),
+                pull_request_id=pull_request.pull_request_id,
+            ),
             params={
-                'csrf_token': csrf_token,
-                'text': 'init',
+                "csrf_token": csrf_token,
+                "text": "init",
             },
             extra_environ=xhr_header,
         )
         assert response.json
         for comment_id in response.json.keys():
-            test_text = 'test'
+            test_text = "test"
             self.app.post(
                 route_path(
-                    'pullrequest_comment_edit',
+                    "pullrequest_comment_edit",
                     repo_name=target_scm_name,
                     pull_request_id=pull_request.pull_request_id,
                     comment_id=comment_id,
                 ),
                 extra_environ=xhr_header,
                 params={
-                    'csrf_token': csrf_token,
-                    'text': test_text,
-                    'version': '0',
+                    "csrf_token": csrf_token,
+                    "text": test_text,
+                    "version": "0",
                 },
-
             )
-            test_text_v2 = 'test_v2'
+            test_text_v2 = "test_v2"
             response = self.app.post(
                 route_path(
-                    'pullrequest_comment_edit',
+                    "pullrequest_comment_edit",
                     repo_name=target_scm_name,
                     pull_request_id=pull_request.pull_request_id,
                     comment_id=comment_id,
                 ),
                 extra_environ=xhr_header,
                 params={
-                    'csrf_token': csrf_token,
-                    'text': test_text_v2,
-                    'version': '0',
+                    "csrf_token": csrf_token,
+                    "text": test_text_v2,
+                    "version": "0",
                 },
                 status=409,
             )
             assert response.status_int == 409
 
-            text_form_db = ChangesetComment.query().filter(
-                ChangesetComment.comment_id == comment_id).first().text
+            text_form_db = (
+                ChangesetComment.query()
+                .filter(ChangesetComment.comment_id == comment_id)
+                .first()
+                .text
+            )
 
             assert test_text == text_form_db
             assert test_text_v2 != text_form_db
 
     def test_comment_and_comment_edit_permissions_forbidden(
-            self, autologin_regular_user, user_regular, user_admin, pr_util,
-            csrf_token, xhr_header):
+        self,
+        autologin_regular_user,
+        user_regular,
+        user_admin,
+        pr_util,
+        csrf_token,
+        xhr_header,
+    ):
         pull_request = pr_util.create_pull_request(
-            author=user_admin.username, enable_notifications=False)
+            author=user_admin.username, enable_notifications=False
+        )
         comment = CommentsModel().create(
-            text='test',
+            text="test",
             repo=pull_request.target_repo.scm_instance().name,
             user=user_admin,
             pull_request=pull_request,
         )
         response = self.app.post(
             route_path(
-                'pullrequest_comment_edit',
+                "pullrequest_comment_edit",
                 repo_name=pull_request.target_repo.scm_instance().name,
                 pull_request_id=pull_request.pull_request_id,
                 comment_id=comment.comment_id,
             ),
             extra_environ=xhr_header,
             params={
-                'csrf_token': csrf_token,
-                'text': 'test_text',
+                "csrf_token": csrf_token,
+                "text": "test_text",
             },
             status=403,
         )
         assert response.status_int == 403
 
     def test_create_pull_request(self, backend, csrf_token):
         commits = [
-            {'message': 'ancestor'},
-            {'message': 'change'},
-            {'message': 'change2'},
+            {"message": "ancestor"},
+            {"message": "change"},
+            {"message": "change2"},
         ]
         commit_ids = backend.create_master_repo(commits)
-        target = backend.create_repo(heads=['ancestor'])
-        source = backend.create_repo(heads=['change2'])
+        target = backend.create_repo(heads=["ancestor"])
+        source = backend.create_repo(heads=["change2"])
 
         response = self.app.post(
-            route_path('pullrequest_create', repo_name=source.repo_name),
+            route_path("pullrequest_create", repo_name=source.repo_name),
             [
-                ('source_repo', source.repo_name),
-                ('source_ref', 'branch:default:' + commit_ids['change2']),
-                ('target_repo', target.repo_name),
-                ('target_ref',  'branch:default:' + commit_ids['ancestor']),
-                ('common_ancestor', commit_ids['ancestor']),
-                ('pullrequest_title', 'Title'),
-                ('pullrequest_desc', 'Description'),
-                ('description_renderer', 'markdown'),
-                ('__start__', 'review_members:sequence'),
-                    ('__start__', 'reviewer:mapping'),
-                        ('user_id', '1'),
-                        ('__start__', 'reasons:sequence'),
-                            ('reason', 'Some reason'),
-                        ('__end__', 'reasons:sequence'),
-                        ('__start__', 'rules:sequence'),
-                        ('__end__', 'rules:sequence'),
-                        ('mandatory', 'False'),
-                    ('__end__', 'reviewer:mapping'),
-                ('__end__', 'review_members:sequence'),
-                ('__start__', 'revisions:sequence'),
-                    ('revisions', commit_ids['change']),
-                    ('revisions', commit_ids['change2']),
-                ('__end__', 'revisions:sequence'),
-                ('user', ''),
-                ('csrf_token', csrf_token),
+                ("source_repo", source.repo_name),
+                ("source_ref", "branch:default:" + commit_ids["change2"]),
+                ("target_repo", target.repo_name),
+                ("target_ref", "branch:default:" + commit_ids["ancestor"]),
+                ("common_ancestor", commit_ids["ancestor"]),
+                ("pullrequest_title", "Title"),
+                ("pullrequest_desc", "Description"),
+                ("description_renderer", "markdown"),
+                ("__start__", "review_members:sequence"),
+                ("__start__", "reviewer:mapping"),
+                ("user_id", "1"),
+                ("__start__", "reasons:sequence"),
+                ("reason", "Some reason"),
+                ("__end__", "reasons:sequence"),
+                ("__start__", "rules:sequence"),
+                ("__end__", "rules:sequence"),
+                ("mandatory", "False"),
+                ("__end__", "reviewer:mapping"),
+                ("__end__", "review_members:sequence"),
+                ("__start__", "revisions:sequence"),
+                ("revisions", commit_ids["change"]),
+                ("revisions", commit_ids["change2"]),
+                ("__end__", "revisions:sequence"),
+                ("user", ""),
+                ("csrf_token", csrf_token),
             ],
-            status=302)
+            status=302,
+        )
 
-        location = response.headers['Location']
-        pull_request_id = location.rsplit('/', 1)[1]
-        assert pull_request_id != 'new'
+        location = response.headers["Location"]
+        pull_request_id = location.rsplit("/", 1)[1]
+        assert pull_request_id != "new"
         pull_request = PullRequest.get(int(pull_request_id))
 
         # check that we have now both revisions
-        assert pull_request.revisions == [commit_ids['change2'], commit_ids['change']]
-        assert pull_request.source_ref == 'branch:default:' + commit_ids['change2']
-        expected_target_ref = 'branch:default:' + commit_ids['ancestor']
+        assert pull_request.revisions == [commit_ids["change2"], commit_ids["change"]]
+        assert pull_request.source_ref == "branch:default:" + commit_ids["change2"]
+        expected_target_ref = "branch:default:" + commit_ids["ancestor"]
         assert pull_request.target_ref == expected_target_ref
 
     def test_reviewer_notifications(self, backend, csrf_token):
-        # We have to use the app.post for this test so it will create the
+        # We have to use the app.post for this test, so it will create the
         # notifications properly with the new PR
         commits = [
-            {'message': 'ancestor',
-             'added': [FileNode(b'file_A', content=b'content_of_ancestor')]},
-            {'message': 'change',
-             'added': [FileNode(b'file_a', content=b'content_of_change')]},
-            {'message': 'change-child'},
-            {'message': 'ancestor-child', 'parents': ['ancestor'],
-             'added': [ FileNode(b'file_B', content=b'content_of_ancestor_child')]},
-            {'message': 'ancestor-child-2'},
+            {
+                "message": "ancestor",
+                "added": [FileNode(b"file_A", content=b"content_of_ancestor")],
+            },
+            {
+                "message": "change",
+                "added": [FileNode(b"file_a", content=b"content_of_change")],
+            },
+            {"message": "change-child"},
+            {
+                "message": "ancestor-child",
+                "parents": ["ancestor"],
+                "branch": "feature",
+                "added": [FileNode(b"file_c", content=b"content_of_ancestor_child")],
+            },
+            {"message": "ancestor-child-2", "branch": "feature"},
         ]
         commit_ids = backend.create_master_repo(commits)
-        target = backend.create_repo(heads=['ancestor-child'])
-        source = backend.create_repo(heads=['change'])
+        target = backend.create_repo(heads=["ancestor-child"])
+        source = backend.create_repo(heads=["change"])
 
         response = self.app.post(
-            route_path('pullrequest_create', repo_name=source.repo_name),
+            route_path("pullrequest_create", repo_name=source.repo_name),
             [
-                ('source_repo', source.repo_name),
-                ('source_ref', 'branch:default:' + commit_ids['change']),
-                ('target_repo', target.repo_name),
-                ('target_ref',  'branch:default:' + commit_ids['ancestor-child']),
-                ('common_ancestor', commit_ids['ancestor']),
-                ('pullrequest_title', 'Title'),
-                ('pullrequest_desc', 'Description'),
-                ('description_renderer', 'markdown'),
-                ('__start__', 'review_members:sequence'),
-                    ('__start__', 'reviewer:mapping'),
-                        ('user_id', '2'),
-                        ('__start__', 'reasons:sequence'),
-                            ('reason', 'Some reason'),
-                        ('__end__', 'reasons:sequence'),
-                        ('__start__', 'rules:sequence'),
-                        ('__end__', 'rules:sequence'),
-                        ('mandatory', 'False'),
-                    ('__end__', 'reviewer:mapping'),
-                ('__end__', 'review_members:sequence'),
-                ('__start__', 'revisions:sequence'),
-                    ('revisions', commit_ids['change']),
-                ('__end__', 'revisions:sequence'),
-                ('user', ''),
-                ('csrf_token', csrf_token),
+                ("source_repo", source.repo_name),
+                ("source_ref", "branch:default:" + commit_ids["change"]),
+                ("target_repo", target.repo_name),
+                ("target_ref", "branch:default:" + commit_ids["ancestor-child"]),
+                ("common_ancestor", commit_ids["ancestor"]),
+                ("pullrequest_title", "Title"),
+                ("pullrequest_desc", "Description"),
+                ("description_renderer", "markdown"),
+                ("__start__", "review_members:sequence"),
+                ("__start__", "reviewer:mapping"),
+                ("user_id", "2"),
+                ("__start__", "reasons:sequence"),
+                ("reason", "Some reason"),
+                ("__end__", "reasons:sequence"),
+                ("__start__", "rules:sequence"),
+                ("__end__", "rules:sequence"),
+                ("mandatory", "False"),
+                ("__end__", "reviewer:mapping"),
+                ("__end__", "review_members:sequence"),
+                ("__start__", "revisions:sequence"),
+                ("revisions", commit_ids["change"]),
+                ("__end__", "revisions:sequence"),
+                ("user", ""),
+                ("csrf_token", csrf_token),
             ],
-            status=302)
+            status=302,
+        )
 
-        location = response.headers['Location']
+        location = response.headers["Location"]
 
-        pull_request_id = location.rsplit('/', 1)[1]
-        assert pull_request_id != 'new'
+        pull_request_id = location.rsplit("/", 1)[1]
+        assert pull_request_id != "new"
         pull_request = PullRequest.get(int(pull_request_id))
 
         # Check that a notification was made
-        notifications = Notification.query()\
-            .filter(Notification.created_by == pull_request.author.user_id,
-                    Notification.type_ == Notification.TYPE_PULL_REQUEST,
-                    Notification.subject.contains(
-                        "requested a pull request review. !%s" % pull_request_id))
+        notifications = Notification.query().filter(
+            Notification.created_by == pull_request.author.user_id,
+            Notification.type_ == Notification.TYPE_PULL_REQUEST,
+            Notification.subject.contains(
+                "requested a pull request review. !%s" % pull_request_id
+            ),
+        )
         assert len(notifications.all()) == 1
 
         # Change reviewers and check that a notification was made
         PullRequestModel().update_reviewers(
-            pull_request.pull_request_id, [
-                (1, [], False, 'reviewer', [])
-            ],
-            pull_request.author)
+            pull_request.pull_request_id,
+            [(1, [], False, "reviewer", [])],
+            pull_request.author,
+        )
         assert len(notifications.all()) == 2
 
     def test_create_pull_request_stores_ancestor_commit_id(self, backend, csrf_token):
         commits = [
-            {'message': 'ancestor',
-             'added': [FileNode(b'file_A', content=b'content_of_ancestor')]},
-            {'message': 'change',
-             'added': [FileNode(b'file_a', content=b'content_of_change')]},
-            {'message': 'change-child'},
-            {'message': 'ancestor-child', 'parents': ['ancestor'],
-             'added': [
-                FileNode(b'file_B', content=b'content_of_ancestor_child')]},
-            {'message': 'ancestor-child-2'},
+            {
+                "message": "ancestor",
+                "added": [FileNode(b"file_A", content=b"content_of_ancestor")],
+            },
+            {
+                "message": "change",
+                "added": [FileNode(b"file_a", content=b"content_of_change")],
+            },
+            {
+                "message": "change-child",
+                "added": [FileNode(b"file_c", content=b"content_of_change_2")],
+            },
+            {
+                "message": "ancestor-child",
+                "parents": ["ancestor"],
+                "branch": "feature",
+                "added": [FileNode(b"file_B", content=b"content_of_ancestor_child")],
+            },
+            {"message": "ancestor-child-2", "branch": "feature"},
         ]
         commit_ids = backend.create_master_repo(commits)
-        target = backend.create_repo(heads=['ancestor-child'])
-        source = backend.create_repo(heads=['change'])
+        target = backend.create_repo(heads=["ancestor-child"])
+        source = backend.create_repo(heads=["change"])
 
         response = self.app.post(
-            route_path('pullrequest_create', repo_name=source.repo_name),
+            route_path("pullrequest_create", repo_name=source.repo_name),
             [
-                ('source_repo', source.repo_name),
-                ('source_ref', 'branch:default:' + commit_ids['change']),
-                ('target_repo', target.repo_name),
-                ('target_ref',  'branch:default:' + commit_ids['ancestor-child']),
-                ('common_ancestor', commit_ids['ancestor']),
-                ('pullrequest_title', 'Title'),
-                ('pullrequest_desc', 'Description'),
-                ('description_renderer', 'markdown'),
-                ('__start__', 'review_members:sequence'),
-                    ('__start__', 'reviewer:mapping'),
-                        ('user_id', '1'),
-                        ('__start__', 'reasons:sequence'),
-                            ('reason', 'Some reason'),
-                        ('__end__', 'reasons:sequence'),
-                        ('__start__', 'rules:sequence'),
-                        ('__end__', 'rules:sequence'),
-                        ('mandatory', 'False'),
-                    ('__end__', 'reviewer:mapping'),
-                ('__end__', 'review_members:sequence'),
-                ('__start__', 'revisions:sequence'),
-                    ('revisions', commit_ids['change']),
-                ('__end__', 'revisions:sequence'),
-                ('user', ''),
-                ('csrf_token', csrf_token),
+                ("source_repo", source.repo_name),
+                ("source_ref", "branch:default:" + commit_ids["change"]),
+                ("target_repo", target.repo_name),
+                ("target_ref", "branch:default:" + commit_ids["ancestor-child"]),
+                ("common_ancestor", commit_ids["ancestor"]),
+                ("pullrequest_title", "Title"),
+                ("pullrequest_desc", "Description"),
+                ("description_renderer", "markdown"),
+                ("__start__", "review_members:sequence"),
+                ("__start__", "reviewer:mapping"),
+                ("user_id", "1"),
+                ("__start__", "reasons:sequence"),
+                ("reason", "Some reason"),
+                ("__end__", "reasons:sequence"),
+                ("__start__", "rules:sequence"),
+                ("__end__", "rules:sequence"),
+                ("mandatory", "False"),
+                ("__end__", "reviewer:mapping"),
+                ("__end__", "review_members:sequence"),
+                ("__start__", "revisions:sequence"),
+                ("revisions", commit_ids["change"]),
+                ("__end__", "revisions:sequence"),
+                ("user", ""),
+                ("csrf_token", csrf_token),
             ],
-            status=302)
+            status=302,
+        )
 
-        location = response.headers['Location']
+        location = response.headers["Location"]
 
-        pull_request_id = location.rsplit('/', 1)[1]
-        assert pull_request_id != 'new'
+        pull_request_id = location.rsplit("/", 1)[1]
+        assert pull_request_id != "new"
         pull_request = PullRequest.get(int(pull_request_id))
 
         # target_ref has to point to the ancestor's commit_id in order to
         # show the correct diff
-        expected_target_ref = 'branch:default:' + commit_ids['ancestor']
+        expected_target_ref = "branch:default:" + commit_ids["ancestor"]
         assert pull_request.target_ref == expected_target_ref
 
         # Check generated diff contents
         response = response.follow()
-        response.mustcontain(no=['content_of_ancestor'])
-        response.mustcontain(no=['content_of_ancestor-child'])
-        response.mustcontain('content_of_change')
+        response.mustcontain(no=["content_of_ancestor"])
+        response.mustcontain(no=["content_of_ancestor-child"])
+        response.mustcontain("content_of_change")
 
     def test_merge_pull_request_enabled(self, pr_util, csrf_token):
         # Clear any previous calls to rcextensions
         rhodecode.EXTENSIONS.calls.clear()
 
-        pull_request = pr_util.create_pull_request(
-            approved=True, mergeable=True)
+        pull_request = pr_util.create_pull_request(approved=True, mergeable=True)
         pull_request_id = pull_request.pull_request_id
-        repo_name = pull_request.target_repo.scm_instance().name,
+        repo_name = (pull_request.target_repo.scm_instance().name,)
 
-        url = route_path('pullrequest_merge',
-                         repo_name=str(repo_name[0]),
-                         pull_request_id=pull_request_id)
-        response = self.app.post(url, params={'csrf_token': csrf_token}).follow()
+        url = route_path(
+            "pullrequest_merge",
+            repo_name=str(repo_name[0]),
+            pull_request_id=pull_request_id,
+        )
+        response = self.app.post(url, params={"csrf_token": csrf_token}).follow()
 
         pull_request = PullRequest.get(pull_request_id)
 
         assert response.status_int == 200
         assert pull_request.is_closed()
-        assert_pull_request_status(
-            pull_request, ChangesetStatus.STATUS_APPROVED)
+        assert_pull_request_status(pull_request, ChangesetStatus.STATUS_APPROVED)
 
         # Check the relevant log entries were added
         user_logs = UserLog.query().order_by(UserLog.user_log_id.desc()).limit(3)
         actions = [log.action for log in user_logs]
         pr_commit_ids = PullRequestModel()._get_commit_ids(pull_request)
         expected_actions = [
-             'repo.pull_request.close',
-             'repo.pull_request.merge',
-             'repo.pull_request.comment.create'
+            "repo.pull_request.close",
+            "repo.pull_request.merge",
+            "repo.pull_request.comment.create",
         ]
         assert actions == expected_actions
 
         user_logs = UserLog.query().order_by(UserLog.user_log_id.desc()).limit(4)
         actions = [log for log in user_logs]
-        assert actions[-1].action == 'user.push'
-        assert actions[-1].action_data['commit_ids'] == pr_commit_ids
+        assert actions[-1].action == "user.push"
+        assert actions[-1].action_data["commit_ids"] == pr_commit_ids
 
         # Check post_push rcextension was really executed
-        push_calls = rhodecode.EXTENSIONS.calls['_push_hook']
+        push_calls = rhodecode.EXTENSIONS.calls["_push_hook"]
         assert len(push_calls) == 1
         unused_last_call_args, last_call_kwargs = push_calls[0]
-        assert last_call_kwargs['action'] == 'push'
-        assert last_call_kwargs['commit_ids'] == pr_commit_ids
+        assert last_call_kwargs["action"] == "push"
+        assert last_call_kwargs["commit_ids"] == pr_commit_ids
 
     def test_merge_pull_request_disabled(self, pr_util, csrf_token):
         pull_request = pr_util.create_pull_request(mergeable=False)
         pull_request_id = pull_request.pull_request_id
         pull_request = PullRequest.get(pull_request_id)
 
         response = self.app.post(
-            route_path('pullrequest_merge',
-                       repo_name=pull_request.target_repo.scm_instance().name,
-                       pull_request_id=pull_request.pull_request_id),
-            params={'csrf_token': csrf_token}).follow()
+            route_path(
+                "pullrequest_merge",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request.pull_request_id,
+            ),
+            params={"csrf_token": csrf_token},
+        ).follow()
 
         assert response.status_int == 200
         response.mustcontain(
-            'Merge is not currently possible because of below failed checks.')
-        response.mustcontain('Server-side pull request merging is disabled.')
+            "Merge is not currently possible because of below failed checks."
+        )
+        response.mustcontain("Server-side pull request merging is disabled.")
 
-    @pytest.mark.skip_backends('svn')
+    @pytest.mark.skip_backends("svn")
     def test_merge_pull_request_not_approved(self, pr_util, csrf_token):
         pull_request = pr_util.create_pull_request(mergeable=True)
         pull_request_id = pull_request.pull_request_id
         repo_name = pull_request.target_repo.scm_instance().name
 
         response = self.app.post(
-            route_path('pullrequest_merge',
-                       repo_name=repo_name, pull_request_id=pull_request_id),
-            params={'csrf_token': csrf_token}).follow()
+            route_path(
+                "pullrequest_merge",
+                repo_name=repo_name,
+                pull_request_id=pull_request_id,
+            ),
+            params={"csrf_token": csrf_token},
+        ).follow()
 
         assert response.status_int == 200
 
         response.mustcontain(
-            'Merge is not currently possible because of below failed checks.')
-        response.mustcontain('Pull request reviewer approval is pending.')
+            "Merge is not currently possible because of below failed checks."
+        )
+        response.mustcontain("Pull request reviewer approval is pending.")
 
     def test_merge_pull_request_renders_failure_reason(
-            self, user_regular, csrf_token, pr_util):
+        self, user_regular, csrf_token, pr_util
+    ):
         pull_request = pr_util.create_pull_request(mergeable=True, approved=True)
         pull_request_id = pull_request.pull_request_id
         repo_name = pull_request.target_repo.scm_instance().name
 
-        merge_resp = MergeResponse(True, False, 'STUB_COMMIT_ID',
-                                   MergeFailureReason.PUSH_FAILED,
-                                   metadata={'target': 'shadow repo',
-                                             'merge_commit': 'xxx'})
+        merge_resp = MergeResponse(
+            True,
+            False,
+            Reference("commit", "STUB_COMMIT_ID", "STUB_COMMIT_ID"),
+            MergeFailureReason.PUSH_FAILED,
+            metadata={"target": "shadow repo", "merge_commit": "xxx"},
+        )
         model_patcher = mock.patch.multiple(
             PullRequestModel,
             merge_repo=mock.Mock(return_value=merge_resp),
-            merge_status=mock.Mock(return_value=(None, True, 'WRONG_MESSAGE')))
+            merge_status=mock.Mock(return_value=(None, True, "WRONG_MESSAGE")),
+        )
 
         with model_patcher:
             response = self.app.post(
-                route_path('pullrequest_merge',
-                           repo_name=repo_name,
-                           pull_request_id=pull_request_id),
-                params={'csrf_token': csrf_token}, status=302)
+                route_path(
+                    "pullrequest_merge",
+                    repo_name=repo_name,
+                    pull_request_id=pull_request_id,
+                ),
+                params={"csrf_token": csrf_token},
+                status=302,
+            )
 
-            merge_resp = MergeResponse(True, True, '', MergeFailureReason.PUSH_FAILED,
-                                       metadata={'target': 'shadow repo',
-                                                 'merge_commit': 'xxx'})
+            merge_resp = MergeResponse(
+                True,
+                True,
+                Reference("commit", "STUB_COMMIT_ID", "STUB_COMMIT_ID"),
+                MergeFailureReason.PUSH_FAILED,
+                metadata={"target": "shadow repo", "merge_commit": "xxx"},
+            )
         assert_session_flash(response, merge_resp.merge_status_message)
 
     def test_update_source_revision(self, backend, csrf_token):
         commits = [
-            {'message': 'ancestor'},
-            {'message': 'change'},
-            {'message': 'change-2'},
+            {"message": "ancestor"},
+            {"message": "change"},
+            {"message": "change-2"},
         ]
         commit_ids = backend.create_master_repo(commits)
-        target = backend.create_repo(heads=['ancestor'])
-        source = backend.create_repo(heads=['change'])
+        target = backend.create_repo(heads=["ancestor"])
+        source = backend.create_repo(heads=["change"])
 
         # create pr from a in source to A in target
         pull_request = PullRequest()
 
         pull_request.source_repo = source
-        pull_request.source_ref = 'branch:{branch}:{commit_id}'.format(
-            branch=backend.default_branch_name, commit_id=commit_ids['change'])
+        pull_request.source_ref = "branch:{branch}:{commit_id}".format(
+            branch=backend.default_branch_name, commit_id=commit_ids["change"]
+        )
 
         pull_request.target_repo = target
-        pull_request.target_ref = 'branch:{branch}:{commit_id}'.format(
-            branch=backend.default_branch_name, commit_id=commit_ids['ancestor'])
+        pull_request.target_ref = "branch:{branch}:{commit_id}".format(
+            branch=backend.default_branch_name, commit_id=commit_ids["ancestor"]
+        )
 
-        pull_request.revisions = [commit_ids['change']]
+        pull_request.revisions = [commit_ids["change"]]
         pull_request.title = "Test"
         pull_request.description = "Description"
         pull_request.author = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         pull_request.pull_request_state = PullRequest.STATE_CREATED
         Session().add(pull_request)
         Session().commit()
         pull_request_id = pull_request.pull_request_id
 
         # source has ancestor - change - change-2
-        backend.pull_heads(source, heads=['change-2'])
+        backend.pull_heads(source, heads=["change-2"])
         target_repo_name = target.repo_name
 
         # update PR
         self.app.post(
-            route_path('pullrequest_update',
-                       repo_name=target_repo_name, pull_request_id=pull_request_id),
-            params={'update_commits': 'true', 'csrf_token': csrf_token})
+            route_path(
+                "pullrequest_update",
+                repo_name=target_repo_name,
+                pull_request_id=pull_request_id,
+            ),
+            params={"update_commits": "true", "csrf_token": csrf_token},
+        )
 
         response = self.app.get(
-            route_path('pullrequest_show',
-                       repo_name=target_repo_name,
-                       pull_request_id=pull_request.pull_request_id))
+            route_path(
+                "pullrequest_show",
+                repo_name=target_repo_name,
+                pull_request_id=pull_request.pull_request_id,
+            )
+        )
 
         assert response.status_int == 200
-        response.mustcontain('Pull request updated to')
-        response.mustcontain('with 1 added, 0 removed commits.')
+        response.mustcontain("Pull request updated to")
+        response.mustcontain("with 1 added, 0 removed commits.")
 
         # check that we have now both revisions
         pull_request = PullRequest.get(pull_request_id)
-        assert pull_request.revisions == [commit_ids['change-2'], commit_ids['change']]
+        assert pull_request.revisions == [commit_ids["change-2"], commit_ids["change"]]
 
     def test_update_target_revision(self, backend, csrf_token):
+        """
+        Checks when we add more commits into a target branch, and update PR
+        """
+
         commits = [
-            {'message': 'ancestor'},
-            {'message': 'change'},
-            {'message': 'ancestor-new', 'parents': ['ancestor']},
-            {'message': 'change-rebased'},
+            {"message": "commit-a"},  # main branch (our PR target)
+            {"message": "commit-b"},  # Initial source
+            {"message": "commit-c"},
+
+            {"message": "commit-a-prime", "branch": "feature", "parents": ["commit-a"]},  # main branch (source)
         ]
+
         commit_ids = backend.create_master_repo(commits)
-        target = backend.create_repo(heads=['ancestor'])
-        source = backend.create_repo(heads=['change'])
+        target = backend.create_repo(heads=["commit-a"])
+        source = backend.create_repo(heads=["commit-b"])
+        target_repo_name = target.repo_name
 
-        # create pr from a in source to A in target
+        # create pr from commit-b to commit-a
         pull_request = PullRequest()
 
-        pull_request.source_repo = source
-        pull_request.source_ref = 'branch:{branch}:{commit_id}'.format(
-            branch=backend.default_branch_name, commit_id=commit_ids['change'])
+        pull_request.target_repo = target
+        pull_request.target_ref = "branch:{branch}:{commit_id}".format(
+            branch=backend.default_branch_name, commit_id=commit_ids["commit-a"]
+        )
 
-        pull_request.target_repo = target
-        pull_request.target_ref = 'branch:{branch}:{commit_id}'.format(
-            branch=backend.default_branch_name, commit_id=commit_ids['ancestor'])
+        pull_request.source_repo = source
+        pull_request.source_ref = "branch:{branch}:{commit_id}".format(
+            branch=backend.default_branch_name, commit_id=commit_ids["commit-b"]
+        )
 
-        pull_request.revisions = [commit_ids['change']]
+        pull_request.revisions = [commit_ids["commit-b"]]
         pull_request.title = "Test"
         pull_request.description = "Description"
         pull_request.author = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         pull_request.pull_request_state = PullRequest.STATE_CREATED
 
         Session().add(pull_request)
         Session().commit()
         pull_request_id = pull_request.pull_request_id
 
-        # target has ancestor - ancestor-new
+        # target - add one commit on top commit-a -> commit-b
+        backend.pull_heads(target, heads=["commit-b"])
+
         # source has ancestor - ancestor-new - change-rebased
-        backend.pull_heads(target, heads=['ancestor-new'])
-        backend.pull_heads(source, heads=['change-rebased'])
-        target_repo_name = target.repo_name
+        backend.pull_heads(source, heads=["commit-c"])
 
         # update PR
-        url = route_path('pullrequest_update',
-                         repo_name=target_repo_name,
-                         pull_request_id=pull_request_id)
-        self.app.post(url,
-                      params={'update_commits': 'true', 'csrf_token': csrf_token},
-                      status=200)
+        url = route_path(
+            "pullrequest_update",
+            repo_name=target_repo_name,
+            pull_request_id=pull_request_id,
+        )
+        self.app.post(
+            url, params={"update_commits": "true", "csrf_token": csrf_token}, status=200
+        )
 
         # check that we have now both revisions
         pull_request = PullRequest.get(pull_request_id)
-        assert pull_request.revisions == [commit_ids['change-rebased']]
-        assert pull_request.target_ref == 'branch:{branch}:{commit_id}'.format(
-            branch=backend.default_branch_name, commit_id=commit_ids['ancestor-new'])
+        assert pull_request.revisions == [commit_ids["commit-c"]]
+        assert pull_request.target_ref == "branch:{branch}:{commit_id}".format(
+            branch=backend.default_branch_name, commit_id=commit_ids["commit-b"]
+        )
 
         response = self.app.get(
-            route_path('pullrequest_show',
-                       repo_name=target_repo_name,
-                       pull_request_id=pull_request.pull_request_id))
+            route_path(
+                "pullrequest_show",
+                repo_name=target_repo_name,
+                pull_request_id=pull_request.pull_request_id,
+            )
+        )
         assert response.status_int == 200
-        response.mustcontain('Pull request updated to')
-        response.mustcontain('with 1 added, 1 removed commits.')
+        response.mustcontain("Pull request updated to")
+        response.mustcontain("with 1 added, 1 removed commits.")
 
-    def test_update_target_revision_with_removal_of_1_commit_git(self, backend_git, csrf_token):
+    def test_update_target_revision_with_removal_of_1_commit_git(
+        self, backend_git, csrf_token
+    ):
         backend = backend_git
         commits = [
-            {'message': 'master-commit-1'},
-            {'message': 'master-commit-2-change-1'},
-            {'message': 'master-commit-3-change-2'},
-
-            {'message': 'feat-commit-1', 'parents': ['master-commit-1']},
-            {'message': 'feat-commit-2'},
+            {"message": "master-commit-1"},
+            {"message": "master-commit-2-change-1"},
+            {"message": "master-commit-3-change-2"},
+            {
+                "message": "feat-commit-1",
+                "parents": ["master-commit-1"],
+                "branch": "feature",
+            },
+            {"message": "feat-commit-2", "branch": "feature"},
         ]
         commit_ids = backend.create_master_repo(commits)
-        target = backend.create_repo(heads=['master-commit-3-change-2'])
-        source = backend.create_repo(heads=['feat-commit-2'])
+        target = backend.create_repo(heads=["master-commit-3-change-2"])
+        source = backend.create_repo(heads=["feat-commit-2"])
 
         # create pr from a in source to A in target
         pull_request = PullRequest()
         pull_request.source_repo = source
 
-        pull_request.source_ref = 'branch:{branch}:{commit_id}'.format(
+        pull_request.source_ref = "branch:{branch}:{commit_id}".format(
             branch=backend.default_branch_name,
-            commit_id=commit_ids['master-commit-3-change-2'])
+            commit_id=commit_ids["master-commit-3-change-2"],
+        )
 
         pull_request.target_repo = target
-        pull_request.target_ref = 'branch:{branch}:{commit_id}'.format(
-            branch=backend.default_branch_name, commit_id=commit_ids['feat-commit-2'])
+        pull_request.target_ref = "branch:{branch}:{commit_id}".format(
+            branch=backend.default_branch_name, commit_id=commit_ids["feat-commit-2"]
+        )
 
         pull_request.revisions = [
-            commit_ids['feat-commit-1'],
-            commit_ids['feat-commit-2']
+            commit_ids["feat-commit-1"],
+            commit_ids["feat-commit-2"],
         ]
         pull_request.title = "Test"
         pull_request.description = "Description"
         pull_request.author = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         pull_request.pull_request_state = PullRequest.STATE_CREATED
         Session().add(pull_request)
         Session().commit()
         pull_request_id = pull_request.pull_request_id
 
         # PR is created, now we simulate a force-push into target,
         # that drops a 2 last commits
         vcsrepo = target.scm_instance()
-        vcsrepo.config.clear_section('hooks')
-        vcsrepo.run_git_command(['reset', '--soft', 'HEAD~2'])
+        vcsrepo.config.clear_section("hooks")
+        vcsrepo.run_git_command(["reset", "--soft", "HEAD~2"])
         target_repo_name = target.repo_name
 
         # update PR
-        url = route_path('pullrequest_update',
-                         repo_name=target_repo_name,
-                         pull_request_id=pull_request_id)
-        self.app.post(url,
-                      params={'update_commits': 'true', 'csrf_token': csrf_token},
-                      status=200)
+        url = route_path(
+            "pullrequest_update",
+            repo_name=target_repo_name,
+            pull_request_id=pull_request_id,
+        )
+        self.app.post(
+            url, params={"update_commits": "true", "csrf_token": csrf_token}, status=200
+        )
 
-        response = self.app.get(route_path('pullrequest_new', repo_name=target_repo_name))
+        response = self.app.get(
+            route_path("pullrequest_new", repo_name=target_repo_name)
+        )
         assert response.status_int == 200
-        response.mustcontain('Pull request updated to')
-        response.mustcontain('with 0 added, 0 removed commits.')
+        response.mustcontain("Pull request updated to")
+        response.mustcontain("with 0 added, 0 removed commits.")
 
-    def test_update_of_ancestor_reference(self, backend, csrf_token):
+    def test_update_pr_ancestor_reference(self, csrf_token, pr_util: PRTestUtility):
         commits = [
-            {'message': 'ancestor'},
-            {'message': 'change'},
-            {'message': 'change-2'},
-            {'message': 'ancestor-new', 'parents': ['ancestor']},
-            {'message': 'change-rebased'},
+            {"message": "ancestor"},
+            {"message": "change"},
+            {"message": "change-2"},
+
+            {"message": "ancestor-new", "parents": ["ancestor"], "branch": "feature"},
+            {"message": "change-rebased", "branch": "feature"},
         ]
-        commit_ids = backend.create_master_repo(commits)
-        target = backend.create_repo(heads=['ancestor'])
-        source = backend.create_repo(heads=['change'])
 
-        # create pr from a in source to A in target
-        pull_request = PullRequest()
-        pull_request.source_repo = source
+        pull_request = pr_util.create_pull_request(
+            commits,
+            target_head="ancestor",
+            source_head="change",
+            revisions=["change"],
+        )
+        pull_request_id = pull_request.pull_request_id
+        target_repo_name = pr_util.target_repository.repo_name
+        commit_ids = pr_util.commit_ids
+
+        assert pull_request.revisions == [commit_ids["change"]]
+        assert list(pull_request.target_repo.scm_instance(cache=False).branches.keys()) == [pr_util.backend.default_branch_name]
+        assert list(pull_request.source_repo.scm_instance(cache=False).branches.keys()) == [pr_util.backend.default_branch_name]
 
-        pull_request.source_ref = 'branch:{branch}:{commit_id}'.format(
-            branch=backend.default_branch_name, commit_id=commit_ids['change'])
-        pull_request.target_repo = target
-        pull_request.target_ref = 'branch:{branch}:{commit_id}'.format(
-            branch=backend.default_branch_name, commit_id=commit_ids['ancestor'])
-        pull_request.revisions = [commit_ids['change']]
-        pull_request.title = "Test"
-        pull_request.description = "Description"
-        pull_request.author = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
-        pull_request.pull_request_state = PullRequest.STATE_CREATED
-        Session().add(pull_request)
+        branch = "feature"
+        pr_util.update_target_repository(head="ancestor-new", do_fetch=True)
+        pr_util.set_pr_target_ref(ref_type="branch", ref_name=branch, ref_commit_id=commit_ids["ancestor-new"])
+
+        pr_util.update_source_repository(head="change-rebased", do_fetch=True)
+        pr_util.set_pr_source_ref(ref_type="branch", ref_name=branch, ref_commit_id=commit_ids["change-rebased"])
+
+        assert list(pull_request.target_repo.scm_instance(cache=False).branches.keys()) == [pr_util.backend.default_branch_name, branch]
+        assert list(pull_request.source_repo.scm_instance(cache=False).branches.keys()) == [pr_util.backend.default_branch_name, branch]
+
+        Session().add(pr_util.pull_request)
         Session().commit()
-        pull_request_id = pull_request.pull_request_id
 
-        # target has ancestor - ancestor-new
-        # source has ancestor - ancestor-new - change-rebased
-        backend.pull_heads(target, heads=['ancestor-new'])
-        backend.pull_heads(source, heads=['change-rebased'])
-        target_repo_name = target.repo_name
+        self.app.post(
+            route_path(
+                "pullrequest_update",
+                repo_name=target_repo_name,
+                pull_request_id=pull_request_id,
+            ),
+            params={"update_commits": "true", "csrf_token": csrf_token, "force_refresh": True},
+            status=200,
+        )
+
 
-        # update PR
-        self.app.post(
-            route_path('pullrequest_update',
-                       repo_name=target_repo_name, pull_request_id=pull_request_id),
-            params={'update_commits': 'true', 'csrf_token': csrf_token},
-            status=200)
+        # response = self.app.get(
+        #     route_path(
+        #         "pullrequest_show", repo_name=target_repo_name, pull_request_id=pull_request_id,
+        #         params={"force_refresh": True}
+        #     ),
+        # )
+        #
+        # response.mustcontain("Pull request updated to")
+        # response.mustcontain("with 1 added, 0 removed commits.")
 
-        # Expect the target reference to be updated correctly
         pull_request = PullRequest.get(pull_request_id)
-        assert pull_request.revisions == [commit_ids['change-rebased']]
-        expected_target_ref = 'branch:{branch}:{commit_id}'.format(
-            branch=backend.default_branch_name,
-            commit_id=commit_ids['ancestor-new'])
-        assert pull_request.target_ref == expected_target_ref
+
+        assert pull_request.target_ref == "branch:{branch}:{commit_id}".format(
+            branch="feature", commit_id=commit_ids["ancestor-new"])
+
+        assert pull_request.revisions == [commit_ids["change-rebased"]]
+
 
     def test_remove_pull_request_branch(self, backend_git, csrf_token):
-        branch_name = 'development'
+        branch_name = "development"
         commits = [
-            {'message': 'initial-commit'},
-            {'message': 'old-feature'},
-            {'message': 'new-feature', 'branch': branch_name},
+            {"message": "initial-commit"},
+            {"message": "old-feature"},
+            {"message": "new-feature", "branch": branch_name},
         ]
         repo = backend_git.create_repo(commits)
         repo_name = repo.repo_name
         commit_ids = backend_git.commit_ids
 
         pull_request = PullRequest()
         pull_request.source_repo = repo
         pull_request.target_repo = repo
-        pull_request.source_ref = 'branch:{branch}:{commit_id}'.format(
-            branch=branch_name, commit_id=commit_ids['new-feature'])
-        pull_request.target_ref = 'branch:{branch}:{commit_id}'.format(
-            branch=backend_git.default_branch_name, commit_id=commit_ids['old-feature'])
-        pull_request.revisions = [commit_ids['new-feature']]
+        pull_request.source_ref = "branch:{branch}:{commit_id}".format(
+            branch=branch_name, commit_id=commit_ids["new-feature"]
+        )
+        pull_request.target_ref = "branch:{branch}:{commit_id}".format(
+            branch=backend_git.default_branch_name, commit_id=commit_ids["old-feature"]
+        )
+        pull_request.revisions = [commit_ids["new-feature"]]
         pull_request.title = "Test"
         pull_request.description = "Description"
         pull_request.author = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         pull_request.pull_request_state = PullRequest.STATE_CREATED
         Session().add(pull_request)
         Session().commit()
 
         pull_request_id = pull_request.pull_request_id
 
         vcs = repo.scm_instance()
-        vcs.remove_ref('refs/heads/{}'.format(branch_name))
+        vcs.remove_ref("refs/heads/{}".format(branch_name))
         # NOTE(marcink): run GC to ensure the commits are gone
         vcs.run_gc()
 
-        response = self.app.get(route_path(
-            'pullrequest_show',
-            repo_name=repo_name,
-            pull_request_id=pull_request_id))
+        response = self.app.get(
+            route_path(
+                "pullrequest_show", repo_name=repo_name, pull_request_id=pull_request_id
+            )
+        )
 
         assert response.status_int == 200
 
         response.assert_response().element_contains(
-            '#changeset_compare_view_content .alert strong',
-            'Missing commits')
+            "#changeset_compare_view_content .alert strong", "Missing commits"
+        )
         response.assert_response().element_contains(
-            '#changeset_compare_view_content .alert',
-            'This pull request cannot be displayed, because one or more'
-            ' commits no longer exist in the source repository.')
+            "#changeset_compare_view_content .alert",
+            "This pull request cannot be displayed, because one or more"
+            " commits no longer exist in the source repository.",
+        )
 
-    def test_strip_commits_from_pull_request(
-            self, backend, pr_util, csrf_token):
+    def test_strip_commits_from_pull_request(self, backend, pr_util):
         commits = [
-            {'message': 'initial-commit'},
-            {'message': 'old-feature'},
-            {'message': 'new-feature', 'parents': ['initial-commit']},
+            {"message": "initial-commit"},
+            {"message": "old-feature"},
+            {"message": "new-feature"},
         ]
         pull_request = pr_util.create_pull_request(
-            commits, target_head='initial-commit', source_head='new-feature',
-            revisions=['new-feature'])
+            commits,
+            target_head="initial-commit",
+            source_head="new-feature",
+            revisions=["new-feature"],
+        )
 
         vcs = pr_util.source_repository.scm_instance()
-        if backend.alias == 'git':
-            vcs.strip(pr_util.commit_ids['new-feature'], branch_name='master')
+        if backend.alias == "git":
+            vcs.strip(pr_util.commit_ids["new-feature"], branch_name=pr_util.backend.default_branch_name)
         else:
-            vcs.strip(pr_util.commit_ids['new-feature'])
+            vcs.strip(pr_util.commit_ids["new-feature"])
 
-        response = self.app.get(route_path(
-            'pullrequest_show',
-            repo_name=pr_util.target_repository.repo_name,
-            pull_request_id=pull_request.pull_request_id))
+        response = self.app.get(
+            route_path(
+                "pullrequest_show",
+                repo_name=pr_util.target_repository.repo_name,
+                pull_request_id=pull_request.pull_request_id,
+            )
+        )
 
         assert response.status_int == 200
 
         response.assert_response().element_contains(
-            '#changeset_compare_view_content .alert strong',
-            'Missing commits')
-        response.assert_response().element_contains(
-            '#changeset_compare_view_content .alert',
-            'This pull request cannot be displayed, because one or more'
-            ' commits no longer exist in the source repository.')
+            "#changeset_compare_view_content .alert strong", "Missing commits"
+        )
         response.assert_response().element_contains(
-            '#update_commits',
-            'Update commits')
+            "#changeset_compare_view_content .alert",
+            "This pull request cannot be displayed, because one or more"
+            " commits no longer exist in the source repository.",
+        )
+        response.assert_response().element_contains("#update_commits", "Update commits")
 
-    def test_strip_commits_and_update(
-            self, backend, pr_util, csrf_token):
+    def test_strip_commits_and_update(self, backend, pr_util, csrf_token):
         commits = [
-            {'message': 'initial-commit'},
-            {'message': 'old-feature'},
-            {'message': 'new-feature', 'parents': ['old-feature']},
+            {"message": "initial-commit"},
+            {"message": "old-feature"},
+            {"message": "new-feature", "parents": ["old-feature"]},
         ]
         pull_request = pr_util.create_pull_request(
-            commits, target_head='old-feature', source_head='new-feature',
-            revisions=['new-feature'], mergeable=True)
+            commits,
+            target_head="old-feature",
+            source_head="new-feature",
+            revisions=["new-feature"],
+            mergeable=True,
+        )
         pr_id = pull_request.pull_request_id
         target_repo_name = pull_request.target_repo.repo_name
 
         vcs = pr_util.source_repository.scm_instance()
-        if backend.alias == 'git':
-            vcs.strip(pr_util.commit_ids['new-feature'], branch_name='master')
+        if backend.alias == "git":
+            vcs.strip(pr_util.commit_ids["new-feature"], branch_name="master")
         else:
-            vcs.strip(pr_util.commit_ids['new-feature'])
+            vcs.strip(pr_util.commit_ids["new-feature"])
 
-        url = route_path('pullrequest_update',
-                         repo_name=target_repo_name,
-                         pull_request_id=pr_id)
-        response = self.app.post(url,
-                                 params={'update_commits': 'true',
-                                         'csrf_token': csrf_token})
+        url = route_path(
+            "pullrequest_update", repo_name=target_repo_name, pull_request_id=pr_id
+        )
+        response = self.app.post(
+            url, params={"update_commits": "true", "csrf_token": csrf_token}
+        )
 
         assert response.status_int == 200
-        assert json.loads(response.body) == json.loads('{"response": true, "redirect_url": null}')
+        assert json.loads(response.body) == json.loads(
+            '{"response": true, "redirect_url": null}'
+        )
 
         # Make sure that after update, it won't raise 500 errors
-        response = self.app.get(route_path(
-            'pullrequest_show',
-            repo_name=target_repo_name,
-            pull_request_id=pr_id))
+        response = self.app.get(
+            route_path(
+                "pullrequest_show", repo_name=target_repo_name, pull_request_id=pr_id
+            )
+        )
 
         assert response.status_int == 200
         response.assert_response().element_contains(
-            '#changeset_compare_view_content .alert strong',
-            'Missing commits')
+            "#changeset_compare_view_content .alert strong", "Missing commits"
+        )
 
     def test_branch_is_a_link(self, pr_util):
         pull_request = pr_util.create_pull_request()
-        pull_request.source_ref = 'branch:origin:1234567890abcdef'
-        pull_request.target_ref = 'branch:target:abcdef1234567890'
+        pull_request.source_ref = "branch:origin:1234567890abcdef"
+        pull_request.target_ref = "branch:target:abcdef1234567890"
         Session().add(pull_request)
         Session().commit()
 
-        response = self.app.get(route_path(
-            'pullrequest_show',
-            repo_name=pull_request.target_repo.scm_instance().name,
-            pull_request_id=pull_request.pull_request_id))
+        response = self.app.get(
+            route_path(
+                "pullrequest_show",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request.pull_request_id,
+            )
+        )
         assert response.status_int == 200
 
-        source = response.assert_response().get_element('.pr-source-info')
+        source = response.assert_response().get_element(".pr-source-info")
         source_parent = source.getparent()
         assert len(source_parent) == 1
 
-        target = response.assert_response().get_element('.pr-target-info')
+        target = response.assert_response().get_element(".pr-target-info")
         target_parent = target.getparent()
         assert len(target_parent) == 1
 
         expected_origin_link = route_path(
-            'repo_commits',
+            "repo_commits",
             repo_name=pull_request.source_repo.scm_instance().name,
-            params=dict(branch='origin'))
+            params=dict(branch="origin"),
+        )
         expected_target_link = route_path(
-            'repo_commits',
+            "repo_commits",
             repo_name=pull_request.target_repo.scm_instance().name,
-            params=dict(branch='target'))
-        assert source_parent.attrib['href'] == expected_origin_link
-        assert target_parent.attrib['href'] == expected_target_link
+            params=dict(branch="target"),
+        )
+        assert source_parent.attrib["href"] == expected_origin_link
+        assert target_parent.attrib["href"] == expected_target_link
 
     def test_bookmark_is_not_a_link(self, pr_util):
         pull_request = pr_util.create_pull_request()
-        pull_request.source_ref = 'bookmark:origin:1234567890abcdef'
-        pull_request.target_ref = 'bookmark:target:abcdef1234567890'
+        pull_request.source_ref = "bookmark:origin:1234567890abcdef"
+        pull_request.target_ref = "bookmark:target:abcdef1234567890"
         Session().add(pull_request)
         Session().commit()
 
-        response = self.app.get(route_path(
-            'pullrequest_show',
-            repo_name=pull_request.target_repo.scm_instance().name,
-            pull_request_id=pull_request.pull_request_id))
+        response = self.app.get(
+            route_path(
+                "pullrequest_show",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request.pull_request_id,
+            )
+        )
         assert response.status_int == 200
 
-        source = response.assert_response().get_element('.pr-source-info')
-        assert source.text.strip() == 'bookmark:origin'
-        assert source.getparent().attrib.get('href') is None
+        source = response.assert_response().get_element(".pr-source-info")
+        assert source.text.strip() == "bookmark:origin"
+        assert source.getparent().attrib.get("href") is None
 
-        target = response.assert_response().get_element('.pr-target-info')
-        assert target.text.strip() == 'bookmark:target'
-        assert target.getparent().attrib.get('href') is None
+        target = response.assert_response().get_element(".pr-target-info")
+        assert target.text.strip() == "bookmark:target"
+        assert target.getparent().attrib.get("href") is None
 
     def test_tag_is_not_a_link(self, pr_util):
         pull_request = pr_util.create_pull_request()
-        pull_request.source_ref = 'tag:origin:1234567890abcdef'
-        pull_request.target_ref = 'tag:target:abcdef1234567890'
+        pull_request.source_ref = "tag:origin:1234567890abcdef"
+        pull_request.target_ref = "tag:target:abcdef1234567890"
         Session().add(pull_request)
         Session().commit()
 
-        response = self.app.get(route_path(
-            'pullrequest_show',
-            repo_name=pull_request.target_repo.scm_instance().name,
-            pull_request_id=pull_request.pull_request_id))
+        response = self.app.get(
+            route_path(
+                "pullrequest_show",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request.pull_request_id,
+            )
+        )
         assert response.status_int == 200
 
-        source = response.assert_response().get_element('.pr-source-info')
-        assert source.text.strip() == 'tag:origin'
-        assert source.getparent().attrib.get('href') is None
+        source = response.assert_response().get_element(".pr-source-info")
+        assert source.text.strip() == "tag:origin"
+        assert source.getparent().attrib.get("href") is None
 
-        target = response.assert_response().get_element('.pr-target-info')
-        assert target.text.strip() == 'tag:target'
-        assert target.getparent().attrib.get('href') is None
+        target = response.assert_response().get_element(".pr-target-info")
+        assert target.text.strip() == "tag:target"
+        assert target.getparent().attrib.get("href") is None
 
-    @pytest.mark.parametrize('mergeable', [True, False])
-    def test_shadow_repository_link(
-            self, mergeable, pr_util, http_host_only_stub):
+    @pytest.mark.parametrize("mergeable", [True, False])
+    def test_shadow_repository_link(self, mergeable, pr_util, http_host_only_stub):
         """
         Check that the pull request summary page displays a link to the shadow
         repository if the pull request is mergeable. If it is not mergeable
         the link should not be displayed.
         """
         pull_request = pr_util.create_pull_request(
-            mergeable=mergeable, enable_notifications=False)
+            mergeable=mergeable, enable_notifications=False
+        )
         target_repo = pull_request.target_repo.scm_instance()
         pr_id = pull_request.pull_request_id
-        shadow_url = '{host}/{repo}/pull-request/{pr_id}/repository'.format(
-            host=http_host_only_stub, repo=target_repo.name, pr_id=pr_id)
+        shadow_url = "{host}/{repo}/pull-request/{pr_id}/repository".format(
+            host=http_host_only_stub, repo=target_repo.name, pr_id=pr_id
+        )
 
-        response = self.app.get(route_path(
-            'pullrequest_show',
-            repo_name=target_repo.name,
-            pull_request_id=pr_id))
+        response = self.app.get(
+            route_path(
+                "pullrequest_show", repo_name=target_repo.name, pull_request_id=pr_id
+            )
+        )
 
         if mergeable:
             response.assert_response().element_value_contains(
-                'input.pr-mergeinfo', shadow_url)
+                "input.pr-mergeinfo", shadow_url
+            )
             response.assert_response().element_value_contains(
-                'input.pr-mergeinfo ', 'pr-merge')
+                "input.pr-mergeinfo ", "pr-merge"
+            )
         else:
-            response.assert_response().no_element_exists('.pr-mergeinfo')
+            response.assert_response().no_element_exists(".pr-mergeinfo")
 
 
[e-mail]('app')
[e-mail]("app")
 @pytest.mark.backends("git", "hg")
 class TestPullrequestsControllerDelete(object):
     def test_pull_request_delete_button_permissions_admin(
-            self, autologin_user, user_admin, pr_util):
+        self, autologin_user, user_admin, pr_util
+    ):
         pull_request = pr_util.create_pull_request(
-            author=user_admin.username, enable_notifications=False)
+            author=user_admin.username, enable_notifications=False
+        )
 
-        response = self.app.get(route_path(
-            'pullrequest_show',
-            repo_name=pull_request.target_repo.scm_instance().name,
-            pull_request_id=pull_request.pull_request_id))
+        response = self.app.get(
+            route_path(
+                "pullrequest_show",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request.pull_request_id,
+            )
+        )
 
         response.mustcontain('id="delete_pullrequest"')
-        response.mustcontain('Confirm to delete this pull request')
+        response.mustcontain("Confirm to delete this pull request")
 
     def test_pull_request_delete_button_permissions_owner(
-            self, autologin_regular_user, user_regular, pr_util):
+        self, autologin_regular_user, user_regular, pr_util
+    ):
         pull_request = pr_util.create_pull_request(
-            author=user_regular.username, enable_notifications=False)
+            author=user_regular.username, enable_notifications=False
+        )
 
-        response = self.app.get(route_path(
-            'pullrequest_show',
-            repo_name=pull_request.target_repo.scm_instance().name,
-            pull_request_id=pull_request.pull_request_id))
+        response = self.app.get(
+            route_path(
+                "pullrequest_show",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request.pull_request_id,
+            )
+        )
 
         response.mustcontain('id="delete_pullrequest"')
-        response.mustcontain('Confirm to delete this pull request')
+        response.mustcontain("Confirm to delete this pull request")
 
     def test_pull_request_delete_button_permissions_forbidden(
-            self, autologin_regular_user, user_regular, user_admin, pr_util):
+        self, autologin_regular_user, user_regular, user_admin, pr_util
+    ):
         pull_request = pr_util.create_pull_request(
-            author=user_admin.username, enable_notifications=False)
+            author=user_admin.username, enable_notifications=False
+        )
 
-        response = self.app.get(route_path(
-            'pullrequest_show',
-            repo_name=pull_request.target_repo.scm_instance().name,
-            pull_request_id=pull_request.pull_request_id))
+        response = self.app.get(
+            route_path(
+                "pullrequest_show",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request.pull_request_id,
+            )
+        )
         response.mustcontain(no=['id="delete_pullrequest"'])
-        response.mustcontain(no=['Confirm to delete this pull request'])
+        response.mustcontain(no=["Confirm to delete this pull request"])
 
     def test_pull_request_delete_button_permissions_can_update_cannot_delete(
-            self, autologin_regular_user, user_regular, user_admin, pr_util,
-            user_util):
-
+        self, autologin_regular_user, user_regular, user_admin, pr_util, user_util
+    ):
         pull_request = pr_util.create_pull_request(
-            author=user_admin.username, enable_notifications=False)
+            author=user_admin.username, enable_notifications=False
+        )
 
         user_util.grant_user_permission_to_repo(
-            pull_request.target_repo, user_regular,
-            'repository.write')
+            pull_request.target_repo, user_regular, "repository.write"
+        )
 
-        response = self.app.get(route_path(
-            'pullrequest_show',
-            repo_name=pull_request.target_repo.scm_instance().name,
-            pull_request_id=pull_request.pull_request_id))
+        response = self.app.get(
+            route_path(
+                "pullrequest_show",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request.pull_request_id,
+            )
+        )
 
         response.mustcontain('id="open_edit_pullrequest"')
         response.mustcontain('id="delete_pullrequest"')
-        response.mustcontain(no=['Confirm to delete this pull request'])
+        response.mustcontain(no=["Confirm to delete this pull request"])
 
     def test_delete_comment_returns_404_if_comment_does_not_exist(
-            self, autologin_user, pr_util, user_admin, csrf_token, xhr_header):
-
+        self, autologin_user, pr_util, user_admin, csrf_token, xhr_header
+    ):
         pull_request = pr_util.create_pull_request(
-            author=user_admin.username, enable_notifications=False)
+            author=user_admin.username, enable_notifications=False
+        )
 
         self.app.post(
             route_path(
-            'pullrequest_comment_delete',
-            repo_name=pull_request.target_repo.scm_instance().name,
-            pull_request_id=pull_request.pull_request_id,
-            comment_id=1024404),
+                "pullrequest_comment_delete",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request.pull_request_id,
+                comment_id=1024404,
+            ),
             extra_environ=xhr_header,
-            params={'csrf_token': csrf_token},
-            status=404
+            params={"csrf_token": csrf_token},
+            status=404,
         )
 
     def test_delete_comment(
-            self, autologin_user, pr_util, user_admin, csrf_token, xhr_header):
-
+        self, autologin_user, pr_util, user_admin, csrf_token, xhr_header
+    ):
         pull_request = pr_util.create_pull_request(
-            author=user_admin.username, enable_notifications=False)
+            author=user_admin.username, enable_notifications=False
+        )
         comment = pr_util.create_comment()
         comment_id = comment.comment_id
 
         response = self.app.post(
             route_path(
-            'pullrequest_comment_delete',
-            repo_name=pull_request.target_repo.scm_instance().name,
-            pull_request_id=pull_request.pull_request_id,
-            comment_id=comment_id),
+                "pullrequest_comment_delete",
+                repo_name=pull_request.target_repo.scm_instance().name,
+                pull_request_id=pull_request.pull_request_id,
+                comment_id=comment_id,
+            ),
             extra_environ=xhr_header,
-            params={'csrf_token': csrf_token},
-            status=200
+            params={"csrf_token": csrf_token},
+            status=200,
         )
-        assert response.text == 'true'
+        assert response.text == "true"
 
-    @pytest.mark.parametrize('url_type', [
-        'pullrequest_new',
-        'pullrequest_create',
-        'pullrequest_update',
-        'pullrequest_merge',
-    ])
+    @pytest.mark.parametrize(
+        "url_type",
+        [
+            "pullrequest_new",
+            "pullrequest_create",
+            "pullrequest_update",
+            "pullrequest_merge",
+        ],
+    )
     def test_pull_request_is_forbidden_on_archived_repo(
-            self, autologin_user, backend, xhr_header, user_util, url_type):
-
+        self, autologin_user, backend, xhr_header, user_util, url_type
+    ):
         # create a temporary repo
         source = user_util.create_repo(repo_type=backend.alias)
         repo_name = source.repo_name
         repo = Repository.get_by_repo_name(repo_name)
         repo.archived = True
         Session().commit()
 
         response = self.app.get(
-            route_path(url_type, repo_name=repo_name, pull_request_id=1), status=302)
+            route_path(url_type, repo_name=repo_name, pull_request_id=1), status=302
+        )
 
-        msg = 'Action not supported for archived repository.'
+        msg = "Action not supported for archived repository."
         assert_session_flash(response, msg)
 
 
 def assert_pull_request_status(pull_request, expected_status):
     status = ChangesetStatusModel().calculated_review_status(pull_request=pull_request)
     assert status == expected_status
 
 
[e-mail]('route', ['pullrequest_new', 'pullrequest_create'])
[e-mail]("route", ["pullrequest_new", "pullrequest_create"])
 @pytest.mark.usefixtures("autologin_user")
 def test_forbidde_to_repo_summary_for_svn_repositories(backend_svn, app, route):
     app.get(route_path(route, repo_name=backend_svn.repo_name), status=404)

rhodecode/model/scm.py

@@ -1,1044 +1,1044 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/
 
 """
 Scm model for RhodeCode
 """
 
 import os.path
 import traceback
 import logging
 import io
 
 from sqlalchemy import func
 from zope.cachedescriptors.property import Lazy as LazyProperty
 
 import rhodecode
 from rhodecode.lib.str_utils import safe_bytes
 from rhodecode.lib.vcs import get_backend
 from rhodecode.lib.vcs.exceptions import RepositoryError, NodeNotChangedError
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.lib.vcs.backends.base import EmptyCommit
 from rhodecode.lib import helpers as h, rc_cache
 from rhodecode.lib.auth import (
     HasRepoPermissionAny, HasRepoGroupPermissionAny,
     HasUserGroupPermissionAny)
 from rhodecode.lib.exceptions import NonRelativePathError, IMCCommitError
 from rhodecode.lib import hooks_utils
 from rhodecode.lib.utils import (
     get_filesystem_repos, make_db_config)
 from rhodecode.lib.str_utils import safe_str
 from rhodecode.lib.system_info import get_system_info
 from rhodecode.model import BaseModel
 from rhodecode.model.db import (
     or_, false, null,
     Repository, CacheKey, UserFollowing, UserLog, User, RepoGroup,
     PullRequest, FileStore)
 from rhodecode.model.settings import VcsSettingsModel
 from rhodecode.model.validation_schema.validators import url_validator, InvalidCloneUrl
 
 log = logging.getLogger(__name__)
 
 
 class UserTemp(object):
     def __init__(self, user_id):
         self.user_id = user_id
 
     def __repr__(self):
         return "<{}('id:{}')>".format(self.__class__.__name__, self.user_id)
 
 
 class RepoTemp(object):
     def __init__(self, repo_id):
         self.repo_id = repo_id
 
     def __repr__(self):
         return "<{}('id:{}')>".format(self.__class__.__name__, self.repo_id)
 
 
 class SimpleCachedRepoList(object):
     """
     Lighter version of of iteration of repos without the scm initialisation,
     and with cache usage
     """
     def __init__(self, db_repo_list, repos_path, order_by=None, perm_set=None):
         self.db_repo_list = db_repo_list
         self.repos_path = repos_path
         self.order_by = order_by
         self.reversed = (order_by or '').startswith('-')
         if not perm_set:
             perm_set = ['repository.read', 'repository.write',
                         'repository.admin']
         self.perm_set = perm_set
 
     def __len__(self):
         return len(self.db_repo_list)
 
     def __repr__(self):
         return '<{} ({})>'.format(self.__class__.__name__, self.__len__())
 
     def __iter__(self):
         for dbr in self.db_repo_list:
             # check permission at this level
             has_perm = HasRepoPermissionAny(*self.perm_set)(
                 dbr.repo_name, 'SimpleCachedRepoList check')
             if not has_perm:
                 continue
 
             tmp_d = {
                 'name': dbr.repo_name,
                 'dbrepo': dbr.get_dict(),
                 'dbrepo_fork': dbr.fork.get_dict() if dbr.fork else {}
             }
             yield tmp_d
 
 
 class _PermCheckIterator(object):
 
     def __init__(
             self, obj_list, obj_attr, perm_set, perm_checker,
             extra_kwargs=None):
         """
         Creates iterator from given list of objects, additionally
         checking permission for them from perm_set var
 
         :param obj_list: list of db objects
         :param obj_attr: attribute of object to pass into perm_checker
         :param perm_set: list of permissions to check
         :param perm_checker: callable to check permissions against
         """
         self.obj_list = obj_list
         self.obj_attr = obj_attr
         self.perm_set = perm_set
         self.perm_checker = perm_checker(*self.perm_set)
         self.extra_kwargs = extra_kwargs or {}
 
     def __len__(self):
         return len(self.obj_list)
 
     def __repr__(self):
         return '<{} ({})>'.format(self.__class__.__name__, self.__len__())
 
     def __iter__(self):
         for db_obj in self.obj_list:
             # check permission at this level
             # NOTE(marcink): the __dict__.get() is ~4x faster then getattr()
             name = db_obj.__dict__.get(self.obj_attr, None)
             if not self.perm_checker(name, self.__class__.__name__, **self.extra_kwargs):
                 continue
 
             yield db_obj
 
 
 class RepoList(_PermCheckIterator):
 
     def __init__(self, db_repo_list, perm_set=None, extra_kwargs=None):
         if not perm_set:
             perm_set = ['repository.read', 'repository.write', 'repository.admin']
 
         super().__init__(
             obj_list=db_repo_list,
             obj_attr='_repo_name', perm_set=perm_set,
             perm_checker=HasRepoPermissionAny,
             extra_kwargs=extra_kwargs)
 
 
 class RepoGroupList(_PermCheckIterator):
 
     def __init__(self, db_repo_group_list, perm_set=None, extra_kwargs=None):
         if not perm_set:
             perm_set = ['group.read', 'group.write', 'group.admin']
 
         super().__init__(
             obj_list=db_repo_group_list,
             obj_attr='_group_name', perm_set=perm_set,
             perm_checker=HasRepoGroupPermissionAny,
             extra_kwargs=extra_kwargs)
 
 
 class UserGroupList(_PermCheckIterator):
 
     def __init__(self, db_user_group_list, perm_set=None, extra_kwargs=None):
         if not perm_set:
             perm_set = ['usergroup.read', 'usergroup.write', 'usergroup.admin']
 
         super().__init__(
             obj_list=db_user_group_list,
             obj_attr='users_group_name', perm_set=perm_set,
             perm_checker=HasUserGroupPermissionAny,
             extra_kwargs=extra_kwargs)
 
 
 class ScmModel(BaseModel):
     """
     Generic Scm Model
     """
 
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
 
         settings_model = VcsSettingsModel(sa=self.sa)
         return settings_model.get_repos_location()
 
     def repo_scan(self, repos_path=None):
         """
         Listing of repositories in given path. This path should not be a
         repository itself. Return a dictionary of repository objects
 
         :param repos_path: path to directory containing repositories
         """
 
         if repos_path is None:
             repos_path = self.repos_path
 
         log.info('scanning for repositories in %s', repos_path)
 
         config = make_db_config()
         config.set('extensions', 'largefiles', '')
         repos = {}
 
         for name, path in get_filesystem_repos(repos_path, recursive=True):
             # name need to be decomposed and put back together using the /
             # since this is internal storage separator for rhodecode
             name = Repository.normalize_repo_name(name)
 
             try:
                 if name in repos:
                     raise RepositoryError('Duplicate repository name %s '
                                           'found in %s' % (name, path))
                 elif path[0] in rhodecode.BACKENDS:
                     backend = get_backend(path[0])
                     repos[name] = backend(path[1], config=config,
                                           with_wire={"cache": False})
             except OSError:
                 continue
             except RepositoryError:
                 log.exception('Failed to create a repo')
                 continue
 
         log.debug('found %s paths with repositories', len(repos))
         return repos
 
     def get_repos(self, all_repos=None, sort_key=None):
         """
         Get all repositories from db and for each repo create it's
         backend instance and fill that backed with information from database
 
         :param all_repos: list of repository names as strings
             give specific repositories list, good for filtering
 
         :param sort_key: initial sorting of repositories
         """
         if all_repos is None:
             all_repos = self.sa.query(Repository)\
                 .filter(Repository.group_id == null())\
                 .order_by(func.lower(Repository.repo_name)).all()
         repo_iter = SimpleCachedRepoList(
             all_repos, repos_path=self.repos_path, order_by=sort_key)
         return repo_iter
 
     def get_repo_groups(self, all_groups=None):
         if all_groups is None:
             all_groups = RepoGroup.query()\
                 .filter(RepoGroup.group_parent_id == null()).all()
         return [x for x in RepoGroupList(all_groups)]
 
     def mark_for_invalidation(self, repo_name, delete=False):
         """
         Mark caches of this repo invalid in the database. `delete` flag
         removes the cache entries
 
         :param repo_name: the repo_name for which caches should be marked
             invalid, or deleted
         :param delete: delete the entry keys instead of setting bool
             flag on them, and also purge caches used by the dogpile
         """
         repo = Repository.get_by_repo_name(repo_name)
 
         if repo:
             invalidation_namespace = CacheKey.REPO_INVALIDATION_NAMESPACE.format(
                 repo_id=repo.repo_id)
             CacheKey.set_invalidate(invalidation_namespace, delete=delete)
 
             repo_id = repo.repo_id
             config = repo._config
             config.set('extensions', 'largefiles', '')
             repo.update_commit_cache(config=config, cs_cache=None)
             if delete:
                 cache_namespace_uid = f'cache_repo.{repo_id}'
                 rc_cache.clear_cache_namespace('cache_repo', cache_namespace_uid, method=rc_cache.CLEAR_INVALIDATE)
 
     def toggle_following_repo(self, follow_repo_id, user_id):
 
         f = self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_repo_id == follow_repo_id)\
             .filter(UserFollowing.user_id == user_id).scalar()
 
         if f is not None:
             try:
                 self.sa.delete(f)
                 return
             except Exception:
                 log.error(traceback.format_exc())
                 raise
 
         try:
             f = UserFollowing()
             f.user_id = user_id
             f.follows_repo_id = follow_repo_id
             self.sa.add(f)
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def toggle_following_user(self, follow_user_id, user_id):
         f = self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_user_id == follow_user_id)\
             .filter(UserFollowing.user_id == user_id).scalar()
 
         if f is not None:
             try:
                 self.sa.delete(f)
                 return
             except Exception:
                 log.error(traceback.format_exc())
                 raise
 
         try:
             f = UserFollowing()
             f.user_id = user_id
             f.follows_user_id = follow_user_id
             self.sa.add(f)
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def is_following_repo(self, repo_name, user_id, cache=False):
         r = self.sa.query(Repository)\
             .filter(Repository.repo_name == repo_name).scalar()
 
         f = self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_repository == r)\
             .filter(UserFollowing.user_id == user_id).scalar()
 
         return f is not None
 
     def is_following_user(self, username, user_id, cache=False):
         u = User.get_by_username(username)
 
         f = self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_user == u)\
             .filter(UserFollowing.user_id == user_id).scalar()
 
         return f is not None
 
     def get_followers(self, repo):
         repo = self._get_repo(repo)
 
         return self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_repository == repo).count()
 
     def get_forks(self, repo):
         repo = self._get_repo(repo)
         return self.sa.query(Repository)\
             .filter(Repository.fork == repo).count()
 
     def get_pull_requests(self, repo):
         repo = self._get_repo(repo)
         return self.sa.query(PullRequest)\
             .filter(PullRequest.target_repo == repo)\
             .filter(PullRequest.status != PullRequest.STATUS_CLOSED).count()
 
     def get_artifacts(self, repo):
         repo = self._get_repo(repo)
         return self.sa.query(FileStore)\
             .filter(FileStore.repo == repo)\
             .filter(or_(FileStore.hidden == null(), FileStore.hidden == false())).count()
 
     def mark_as_fork(self, repo, fork, user):
         repo = self._get_repo(repo)
         fork = self._get_repo(fork)
         if fork and repo.repo_id == fork.repo_id:
             raise Exception("Cannot set repository as fork of itself")
 
         if fork and repo.repo_type != fork.repo_type:
             raise RepositoryError(
                 "Cannot set repository as fork of repository with other type")
 
         repo.fork = fork
         self.sa.add(repo)
         return repo
 
     def pull_changes(self, repo, username, remote_uri=None, validate_uri=True):
         dbrepo = self._get_repo(repo)
         remote_uri = remote_uri or dbrepo.clone_uri
         if not remote_uri:
             raise Exception("This repository doesn't have a clone uri")
 
         repo = dbrepo.scm_instance(cache=False)
         repo.config.clear_section('hooks')
 
         try:
             # NOTE(marcink): add extra validation so we skip invalid urls
             # this is due this tasks can be executed via scheduler without
             # proper validation of remote_uri
             if validate_uri:
                 config = make_db_config(clear_session=False)
                 url_validator(remote_uri, dbrepo.repo_type, config)
         except InvalidCloneUrl:
             raise
 
         repo_name = dbrepo.repo_name
         try:
             # TODO: we need to make sure those operations call proper hooks !
             repo.fetch(remote_uri)
 
             self.mark_for_invalidation(repo_name)
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def push_changes(self, repo, username, remote_uri=None, validate_uri=True):
         dbrepo = self._get_repo(repo)
         remote_uri = remote_uri or dbrepo.push_uri
         if not remote_uri:
             raise Exception("This repository doesn't have a clone uri")
 
         repo = dbrepo.scm_instance(cache=False)
         repo.config.clear_section('hooks')
 
         try:
             # NOTE(marcink): add extra validation so we skip invalid urls
             # this is due this tasks can be executed via scheduler without
             # proper validation of remote_uri
             if validate_uri:
                 config = make_db_config(clear_session=False)
                 url_validator(remote_uri, dbrepo.repo_type, config)
         except InvalidCloneUrl:
             raise
 
         try:
             repo.push(remote_uri)
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def commit_change(self, repo, repo_name, commit, user, author, message,
-                      content: bytes, f_path: bytes):
+                      content: bytes, f_path: bytes, branch: str = None):
         """
         Commits changes
         """
         user = self._get_user(user)
 
         # message and author needs to be unicode
         # proper backend should then translate that into required type
         message = safe_str(message)
         author = safe_str(author)
         imc = repo.in_memory_commit
         imc.change(FileNode(f_path, content, mode=commit.get_file_mode(f_path)))
         try:
             # TODO: handle pre-push action !
             tip = imc.commit(
                 message=message, author=author, parents=[commit],
-                branch=commit.branch)
+                branch=branch or commit.branch)
         except Exception as e:
             log.error(traceback.format_exc())
             raise IMCCommitError(str(e))
         finally:
             # always clear caches, if commit fails we want fresh object also
             self.mark_for_invalidation(repo_name)
 
         # We trigger the post-push action
         hooks_utils.trigger_post_push_hook(
             username=user.username, action='push_local', hook_type='post_push',
             repo_name=repo_name, repo_type=repo.alias, commit_ids=[tip.raw_id])
         return tip
 
     def _sanitize_path(self, f_path: bytes):
         if f_path.startswith(b'/') or f_path.startswith(b'./') or b'../' in f_path:
             raise NonRelativePathError(b'%b is not an relative path' % f_path)
         if f_path:
             f_path = os.path.normpath(f_path)
         return f_path
 
     def get_dirnode_metadata(self, request, commit, dir_node):
         if not dir_node.is_dir():
             return []
 
         data = []
         for node in dir_node:
             if not node.is_file():
                 # we skip file-nodes
                 continue
 
             last_commit = node.last_commit
             last_commit_date = last_commit.date
             data.append({
                 'name': node.name,
                 'size': h.format_byte_size_binary(node.size),
                 'modified_at': h.format_date(last_commit_date),
                 'modified_ts': last_commit_date.isoformat(),
                 'revision': last_commit.revision,
                 'short_id': last_commit.short_id,
                 'message': h.escape(last_commit.message),
                 'author': h.escape(last_commit.author),
                 'user_profile': h.gravatar_with_user(
                     request, last_commit.author),
             })
 
         return data
 
     def get_nodes(self, repo_name, commit_id, root_path='/', flat=True,
                   extended_info=False, content=False, max_file_bytes=None):
         """
         recursive walk in root dir and return a set of all path in that dir
         based on repository walk function
 
         :param repo_name: name of repository
         :param commit_id: commit id for which to list nodes
         :param root_path: root path to list
         :param flat: return as a list, if False returns a dict with description
         :param extended_info: show additional info such as md5, binary, size etc
         :param content: add nodes content to the return data
         :param max_file_bytes: will not return file contents over this limit
 
         """
         _files = list()
         _dirs = list()
 
         try:
             _repo = self._get_repo(repo_name)
             commit = _repo.scm_instance().get_commit(commit_id=commit_id)
             root_path = root_path.lstrip('/')
 
             # get RootNode, inject pre-load options before walking
             top_node = commit.get_node(root_path)
             extended_info_pre_load = []
             if extended_info:
                 extended_info_pre_load += ['md5']
             top_node.default_pre_load = ['is_binary', 'size'] + extended_info_pre_load
 
             for __, dirs, files in commit.walk(top_node):
 
                 for f in files:
                     _content = None
                     _data = f_name = f.str_path
 
                     if not flat:
                         _data = {
                             "name": h.escape(f_name),
                             "type": "file",
                             }
                         if extended_info:
                             _data.update({
                                 "md5": f.md5,
                                 "binary": f.is_binary,
                                 "size": f.size,
                                 "extension": f.extension,
                                 "mimetype": f.mimetype,
                                 "lines": f.lines()[0]
                             })
 
                         if content:
                             over_size_limit = (max_file_bytes is not None
                                                and f.size > max_file_bytes)
                             full_content = None
                             if not f.is_binary and not over_size_limit:
                                 full_content = f.str_content
 
                             _data.update({
                                 "content": full_content,
                             })
                     _files.append(_data)
 
                 for d in dirs:
                     _data = d_name = d.str_path
                     if not flat:
                         _data = {
                             "name": h.escape(d_name),
                             "type": "dir",
                             }
                     if extended_info:
                         _data.update({
                             "md5": "",
                             "binary": False,
                             "size": 0,
                             "extension": "",
                         })
                     if content:
                         _data.update({
                             "content": None
                         })
                     _dirs.append(_data)
         except RepositoryError:
             log.exception("Exception in get_nodes")
             raise
 
         return _dirs, _files
 
     def get_quick_filter_nodes(self, repo_name, commit_id, root_path='/'):
         """
         Generate files for quick filter in files view
         """
 
         _files = list()
         _dirs = list()
         try:
             _repo = self._get_repo(repo_name)
             commit = _repo.scm_instance().get_commit(commit_id=commit_id)
             root_path = root_path.lstrip('/')
 
             top_node = commit.get_node(root_path)
             top_node.default_pre_load = []
 
             for __, dirs, files in commit.walk(top_node):
                 for f in files:
 
                     _data = {
                         "name": h.escape(f.str_path),
                         "type": "file",
                         }
 
                     _files.append(_data)
 
                 for d in dirs:
 
                     _data = {
                         "name": h.escape(d.str_path),
                         "type": "dir",
                         }
 
                     _dirs.append(_data)
         except RepositoryError:
             log.exception("Exception in get_quick_filter_nodes")
             raise
 
         return _dirs, _files
 
     def get_node(self, repo_name, commit_id, file_path,
                  extended_info=False, content=False, max_file_bytes=None, cache=True):
         """
         retrieve single node from commit
         """
 
         try:
 
             _repo = self._get_repo(repo_name)
             commit = _repo.scm_instance().get_commit(commit_id=commit_id)
 
             file_node = commit.get_node(file_path)
             if file_node.is_dir():
                 raise RepositoryError('The given path is a directory')
 
             _content = None
             f_name = file_node.str_path
 
             file_data = {
                 "name": h.escape(f_name),
                 "type": "file",
             }
 
             if extended_info:
                 file_data.update({
                     "extension": file_node.extension,
                     "mimetype": file_node.mimetype,
                 })
 
                 if cache:
                     md5 = file_node.md5
                     is_binary = file_node.is_binary
                     size = file_node.size
                 else:
                     is_binary, md5, size, _content = file_node.metadata_uncached()
 
                 file_data.update({
                     "md5": md5,
                     "binary": is_binary,
                     "size": size,
                 })
 
             if content and cache:
                 # get content + cache
                 size = file_node.size
                 over_size_limit = (max_file_bytes is not None and size > max_file_bytes)
                 full_content = None
                 all_lines = 0
                 if not file_node.is_binary and not over_size_limit:
                     full_content = safe_str(file_node.content)
                     all_lines, empty_lines = file_node.count_lines(full_content)
 
                 file_data.update({
                     "content": full_content,
                     "lines": all_lines
                 })
             elif content:
                 # get content *without* cache
                 if _content is None:
                     is_binary, md5, size, _content = file_node.metadata_uncached()
 
                 over_size_limit = (max_file_bytes is not None and size > max_file_bytes)
                 full_content = None
                 all_lines = 0
                 if not is_binary and not over_size_limit:
                     full_content = safe_str(_content)
                     all_lines, empty_lines = file_node.count_lines(full_content)
 
                 file_data.update({
                     "content": full_content,
                     "lines": all_lines
                 })
 
         except RepositoryError:
             log.exception("Exception in get_node")
             raise
 
         return file_data
 
     def get_fts_data(self, repo_name, commit_id, root_path='/'):
         """
         Fetch node tree for usage in full text search
         """
 
         tree_info = list()
 
         try:
             _repo = self._get_repo(repo_name)
             commit = _repo.scm_instance().get_commit(commit_id=commit_id)
             root_path = root_path.lstrip('/')
             top_node = commit.get_node(root_path)
             top_node.default_pre_load = []
 
             for __, dirs, files in commit.walk(top_node):
 
                 for f in files:
                     is_binary, md5, size, _content = f.metadata_uncached()
                     _data = {
                         "name": f.str_path,
                         "md5": md5,
                         "extension": f.extension,
                         "binary": is_binary,
                         "size": size
                     }
 
                     tree_info.append(_data)
 
         except RepositoryError:
             log.exception("Exception in get_nodes")
             raise
 
         return tree_info
 
     def create_nodes(self, user, repo, message, nodes, parent_commit=None,
                      author=None, trigger_push_hook=True):
         """
         Commits given multiple nodes into repo
 
         :param user: RhodeCode User object or user_id, the commiter
         :param repo: RhodeCode Repository object
         :param message: commit message
         :param nodes: mapping {filename:{'content':content},...}
         :param parent_commit: parent commit, can be empty than it's
            initial commit
         :param author: author of commit, cna be different that commiter
            only for git
         :param trigger_push_hook: trigger push hooks
 
         :returns: new committed commit
         """
 
         user = self._get_user(user)
         scm_instance = repo.scm_instance(cache=False)
 
         message = safe_str(message)
         commiter = user.full_contact
         author = safe_str(author) if author else commiter
 
         imc = scm_instance.in_memory_commit
 
         if not parent_commit:
             parent_commit = EmptyCommit(alias=scm_instance.alias)
 
         if isinstance(parent_commit, EmptyCommit):
             # EmptyCommit means we're editing empty repository
             parents = None
         else:
             parents = [parent_commit]
 
         upload_file_types = (io.BytesIO, io.BufferedRandom)
         processed_nodes = []
         for filename, content_dict in nodes.items():
             if not isinstance(filename, bytes):
                 raise ValueError(f'filename key in nodes needs to be bytes , or {upload_file_types}')
             content = content_dict['content']
             if not isinstance(content, upload_file_types + (bytes,)):
                 raise ValueError('content key value in nodes needs to be bytes')
 
         for f_path in nodes:
             f_path = self._sanitize_path(f_path)
             content = nodes[f_path]['content']
 
             # decoding here will force that we have proper encoded values
             # in any other case this will throw exceptions and deny commit
 
             if isinstance(content, bytes):
                 pass
             elif isinstance(content, upload_file_types):
                 content = content.read()
             else:
                 raise Exception(f'Content is of unrecognized type {type(content)}, expected {upload_file_types}')
             processed_nodes.append((f_path, content))
 
         # add multiple nodes
         for path, content in processed_nodes:
             imc.add(FileNode(path, content=content))
 
         # TODO: handle pre push scenario
         tip = imc.commit(message=message,
                          author=author,
                          parents=parents,
                          branch=parent_commit.branch)
 
         self.mark_for_invalidation(repo.repo_name)
         if trigger_push_hook:
             hooks_utils.trigger_post_push_hook(
                 username=user.username, action='push_local',
                 repo_name=repo.repo_name, repo_type=scm_instance.alias,
                 hook_type='post_push',
                 commit_ids=[tip.raw_id])
         return tip
 
     def update_nodes(self, user, repo, message, nodes, parent_commit=None,
                      author=None, trigger_push_hook=True):
         user = self._get_user(user)
         scm_instance = repo.scm_instance(cache=False)
 
         message = safe_str(message)
         commiter = user.full_contact
         author = safe_str(author) if author else commiter
 
         imc = scm_instance.in_memory_commit
 
         if not parent_commit:
             parent_commit = EmptyCommit(alias=scm_instance.alias)
 
         if isinstance(parent_commit, EmptyCommit):
             # EmptyCommit means we we're editing empty repository
             parents = None
         else:
             parents = [parent_commit]
 
         # add multiple nodes
         for _filename, data in nodes.items():
             # new filename, can be renamed from the old one, also sanitaze
             # the path for any hack around relative paths like ../../ etc.
             filename = self._sanitize_path(data['filename'])
             old_filename = self._sanitize_path(_filename)
             content = data['content']
             file_mode = data.get('mode')
             filenode = FileNode(old_filename, content=content, mode=file_mode)
             op = data['op']
             if op == 'add':
                 imc.add(filenode)
             elif op == 'del':
                 imc.remove(filenode)
             elif op == 'mod':
                 if filename != old_filename:
                     # TODO: handle renames more efficient, needs vcs lib changes
                     imc.remove(filenode)
                     imc.add(FileNode(filename, content=content, mode=file_mode))
                 else:
                     imc.change(filenode)
 
         try:
             # TODO: handle pre push scenario commit changes
             tip = imc.commit(message=message,
                              author=author,
                              parents=parents,
                              branch=parent_commit.branch)
         except NodeNotChangedError:
             raise
         except Exception as e:
             log.exception("Unexpected exception during call to imc.commit")
             raise IMCCommitError(str(e))
         finally:
             # always clear caches, if commit fails we want fresh object also
             self.mark_for_invalidation(repo.repo_name)
 
         if trigger_push_hook:
             hooks_utils.trigger_post_push_hook(
                 username=user.username, action='push_local', hook_type='post_push',
                 repo_name=repo.repo_name, repo_type=scm_instance.alias,
                 commit_ids=[tip.raw_id])
 
         return tip
 
     def delete_nodes(self, user, repo, message, nodes, parent_commit=None,
                      author=None, trigger_push_hook=True):
         """
         Deletes given multiple nodes into `repo`
 
         :param user: RhodeCode User object or user_id, the committer
         :param repo: RhodeCode Repository object
         :param message: commit message
         :param nodes: mapping {filename:{'content':content},...}
         :param parent_commit: parent commit, can be empty than it's initial
            commit
         :param author: author of commit, cna be different that commiter only
            for git
         :param trigger_push_hook: trigger push hooks
 
         :returns: new commit after deletion
         """
 
         user = self._get_user(user)
         scm_instance = repo.scm_instance(cache=False)
 
         processed_nodes = []
         for f_path in nodes:
             f_path = self._sanitize_path(f_path)
             # content can be empty but for compatibility it allows same dicts
             # structure as add_nodes
             content = nodes[f_path].get('content')
             processed_nodes.append((safe_bytes(f_path), content))
 
         message = safe_str(message)
         commiter = user.full_contact
         author = safe_str(author) if author else commiter
 
         imc = scm_instance.in_memory_commit
 
         if not parent_commit:
             parent_commit = EmptyCommit(alias=scm_instance.alias)
 
         if isinstance(parent_commit, EmptyCommit):
             # EmptyCommit means we we're editing empty repository
             parents = None
         else:
             parents = [parent_commit]
         # add multiple nodes
         for path, content in processed_nodes:
             imc.remove(FileNode(path, content=content))
 
         # TODO: handle pre push scenario
         tip = imc.commit(message=message,
                          author=author,
                          parents=parents,
                          branch=parent_commit.branch)
 
         self.mark_for_invalidation(repo.repo_name)
         if trigger_push_hook:
             hooks_utils.trigger_post_push_hook(
                 username=user.username, action='push_local', hook_type='post_push',
                 repo_name=repo.repo_name, repo_type=scm_instance.alias,
                 commit_ids=[tip.raw_id])
         return tip
 
     def strip(self, repo, commit_id, branch):
         scm_instance = repo.scm_instance(cache=False)
         scm_instance.config.clear_section('hooks')
         scm_instance.strip(commit_id, branch)
         self.mark_for_invalidation(repo.repo_name)
 
     def get_unread_journal(self):
         return self.sa.query(UserLog).count()
 
     @classmethod
     def backend_landing_ref(cls, repo_type):
         """
         Return a default landing ref based on a repository type.
         """
 
         landing_ref = {
             'hg': ('branch:default', 'default'),
             'git': ('branch:master', 'master'),
             'svn': ('rev:tip', 'latest tip'),
             'default': ('rev:tip', 'latest tip'),
         }
 
         return landing_ref.get(repo_type) or landing_ref['default']
 
     def get_repo_landing_revs(self, translator, repo=None):
         """
         Generates select option with tags branches and bookmarks (for hg only)
         grouped by type
 
         :param repo:
         """
         from rhodecode.lib.vcs.backends.git import GitRepository
 
         _ = translator
         repo = self._get_repo(repo)
 
         if repo:
             repo_type = repo.repo_type
         else:
             repo_type = 'default'
 
         default_landing_ref, landing_ref_lbl = self.backend_landing_ref(repo_type)
 
         default_ref_options = [
             [default_landing_ref, landing_ref_lbl]
         ]
         default_choices = [
             default_landing_ref
         ]
 
         if not repo:
             # presented at NEW repo creation
             return default_choices, default_ref_options
 
         repo = repo.scm_instance()
 
         ref_options = [(default_landing_ref, landing_ref_lbl)]
         choices = [default_landing_ref]
 
         # branches
         branch_group = [(f'branch:{safe_str(b)}', safe_str(b)) for b in repo.branches]
         if not branch_group:
             # new repo, or without maybe a branch?
             branch_group = default_ref_options
 
         branches_group = (branch_group, _("Branches"))
         ref_options.append(branches_group)
         choices.extend([x[0] for x in branches_group[0]])
 
         # bookmarks for HG
         if repo.alias == 'hg':
             bookmarks_group = (
                 [(f'book:{safe_str(b)}', safe_str(b))
                     for b in repo.bookmarks],
                 _("Bookmarks"))
             ref_options.append(bookmarks_group)
             choices.extend([x[0] for x in bookmarks_group[0]])
 
         # tags
         tags_group = (
             [(f'tag:{safe_str(t)}', safe_str(t))
                 for t in repo.tags],
             _("Tags"))
         ref_options.append(tags_group)
         choices.extend([x[0] for x in tags_group[0]])
 
         return choices, ref_options
 
     def get_server_info(self, environ=None):
         server_info = get_system_info(environ)
         return server_info

rhodecode/tests/fixture_mods/fixture_utils.py

@@ -1,1735 +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.pyramid_utils import get_app_config
     from rhodecode.config.middleware import make_pyramid_app
 
     log.info("Using the RhodeCode configuration:{}".format(ini_config))
     pyramid.paster.setup_logging(ini_config)
 
     settings = get_app_config(ini_config)
     app = make_pyramid_app({'__file__': ini_config}, **settings)
 
     return app
 
 
 @pytest.fixture(scope='function')
 def app(request, config_stub, baseapp, http_environ):
     app = CustomTestApp(
         baseapp,
         extra_environ=http_environ)
     if request.cls:
         request.cls.app = app
     return app
 
 
 @pytest.fixture(scope='session')
 def app_settings(baseapp, ini_config):
     """
     Settings dictionary used to create the app.
 
     Parses the ini file and passes the result through the sanitize and apply
     defaults mechanism in `rhodecode.config.middleware`.
     """
     return baseapp.config.get_settings()
 
 
 @pytest.fixture(scope='session')
 def db_connection(ini_settings):
     # Initialize the database connection.
     config_utils.initialize_database(ini_settings)
 
 
 LoginData = collections.namedtuple('LoginData', ('csrf_token', 'user'))
 
 
 def _autologin_user(app, *args):
     session = login_user_session(app, *args)
     csrf_token = rhodecode.lib.auth.get_csrf_token(session)
     return LoginData(csrf_token, session['rhodecode_user'])
 
 
 @pytest.fixture()
 def autologin_user(app):
     """
     Utility fixture which makes sure that the admin user is logged in
     """
     return _autologin_user(app)
 
 
 @pytest.fixture()
 def autologin_regular_user(app):
     """
     Utility fixture which makes sure that the regular user is logged in
     """
     return _autologin_user(
         app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
 
 
 @pytest.fixture(scope='function')
 def csrf_token(request, autologin_user):
     return autologin_user.csrf_token
 
 
 @pytest.fixture(scope='function')
 def xhr_header(request):
     return {'HTTP_X_REQUESTED_WITH': 'XMLHttpRequest'}
 
 
 @pytest.fixture()
 def real_crypto_backend(monkeypatch):
     """
     Switch the production crypto backend on for this test.
 
     During the test run the crypto backend is replaced with a faster
     implementation based on the MD5 algorithm.
     """
     monkeypatch.setattr(rhodecode, 'is_test', False)
 
 
 @pytest.fixture(scope='class')
 def index_location(request, baseapp):
     index_location = baseapp.config.get_settings()['search.location']
     if request.cls:
         request.cls.index_location = index_location
     return index_location
 
 
 @pytest.fixture(scope='session', autouse=True)
 def tests_tmp_path(request):
     """
     Create temporary directory to be used during the test session.
     """
     if not os.path.exists(TESTS_TMP_PATH):
         os.makedirs(TESTS_TMP_PATH)
 
     if not request.config.getoption('--keep-tmp-path'):
         @request.addfinalizer
         def remove_tmp_path():
             shutil.rmtree(TESTS_TMP_PATH)
 
     return TESTS_TMP_PATH
 
 
 @pytest.fixture()
 def test_repo_group(request):
     """
     Create a temporary repository group, and destroy it after
     usage automatically
     """
     fixture = Fixture()
     repogroupid = 'test_repo_group_%s' % str(time.time()).replace('.', '')
     repo_group = fixture.create_repo_group(repogroupid)
 
     def _cleanup():
         fixture.destroy_repo_group(repogroupid)
 
     request.addfinalizer(_cleanup)
     return repo_group
 
 
 @pytest.fixture()
 def test_user_group(request):
     """
     Create a temporary user group, and destroy it after
     usage automatically
     """
     fixture = Fixture()
     usergroupid = 'test_user_group_%s' % str(time.time()).replace('.', '')
     user_group = fixture.create_user_group(usergroupid)
 
     def _cleanup():
         fixture.destroy_user_group(user_group)
 
     request.addfinalizer(_cleanup)
     return user_group
 
 
 @pytest.fixture(scope='session')
 def test_repo(request):
     container = TestRepoContainer()
     request.addfinalizer(container._cleanup)
     return container
 
 
 class TestRepoContainer(object):
     """
     Container for test repositories which are used read only.
 
     Repositories will be created on demand and re-used during the lifetime
     of this object.
 
     Usage to get the svn test repository "minimal"::
 
        test_repo = TestContainer()
        repo = test_repo('minimal', 'svn')
 
     """
 
     dump_extractors = {
         'git': utils.extract_git_repo_from_dump,
         'hg': utils.extract_hg_repo_from_dump,
         'svn': utils.extract_svn_repo_from_dump,
     }
 
     def __init__(self):
         self._cleanup_repos = []
         self._fixture = Fixture()
         self._repos = {}
 
     def __call__(self, dump_name, backend_alias, config=None):
         key = (dump_name, backend_alias)
         if key not in self._repos:
             repo = self._create_repo(dump_name, backend_alias, config)
             self._repos[key] = repo.repo_id
         return Repository.get(self._repos[key])
 
     def _create_repo(self, dump_name, backend_alias, config):
         repo_name = '%s-%s' % (backend_alias, dump_name)
         backend = 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': 'Commit %s of %s' % (x, self.repo_name)}
+            {'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):
+    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 u"%s_%s" % (
+        return "%s_%s" % (
             self.invalid_repo_name.sub('_', self._test_name), len(self._cleanup_repos))
 
-    def ensure_file(self, filename, content='Test content\n'):
+    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:
-                # TODO: mikhail: do more special chars replacements
-                ref_name = 'refs/test-refs/{}'.format(
-                    message.replace(' ', ''))
+                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 in refs:
-            repo.set_refs(ref_name, refs[ref_name])
+        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
-    commit = None
 
     for idx, commit in enumerate(commits):
-        message = str(commit.get('message', 'Commit %s' % idx))
+        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=u"Test", description=u"Description"):
+            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):
-        reference = '%s:%s:%s' % (
-            'branch',
-            self.backend.default_branch_name,
-            self.commit_ids[commit_message])
+    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):
+    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.source_repository, heads=heads)
+        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=u"Test comment",
+            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=u"Test comment",
+            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/models/test_scm.py

@@ -1,189 +1,189 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 pytest
 from mock import Mock, patch, DEFAULT
 
 import rhodecode
 from rhodecode.model import db, scm
 
 
 def test_scm_instance_config(backend):
     repo = backend.create_repo()
     with patch.multiple('rhodecode.model.db.Repository',
                         _get_instance=DEFAULT,
                         _get_instance_cached=DEFAULT) as mocks:
 
         repo.scm_instance()
         mocks['_get_instance'].assert_called_with(
             config=None, cache=False)
 
         repo.scm_instance(vcs_full_cache=False)
         mocks['_get_instance'].assert_called_with(
             config=None, cache=False)
 
         repo.scm_instance(vcs_full_cache=True)
         mocks['_get_instance_cached'].assert_called()
 
 
 def test_get_instance_config(backend):
     repo = backend.create_repo()
     vcs_class = Mock()
     with patch.multiple('rhodecode.lib.vcs.backends',
                         get_scm=DEFAULT,
                         get_backend=DEFAULT) as mocks:
         mocks['get_scm'].return_value = backend.alias
         mocks['get_backend'].return_value = vcs_class
         with patch('rhodecode.model.db.Repository._config') as config_mock:
             repo._get_instance()
             vcs_class.assert_called_with(
                 repo_path=repo.repo_full_path, config=config_mock,
                 create=False, with_wire={'cache': True, 'repo_state_uid': None})
 
         new_config = {'override': 'old_config'}
         repo._get_instance(config=new_config)
         vcs_class.assert_called_with(
             repo_path=repo.repo_full_path, config=new_config, create=False,
             with_wire={'cache': True, 'repo_state_uid': None})
 
 
 def test_mark_for_invalidation_config(backend):
     repo = backend.create_repo()
     with patch('rhodecode.model.db.Repository.update_commit_cache') as _mock:
         scm.ScmModel().mark_for_invalidation(repo.repo_name)
         _, kwargs = _mock.call_args
         assert kwargs['config'].__dict__ == repo._config.__dict__
 
 
 def test_mark_for_invalidation_with_delete_updates_last_commit(backend):
     commits = [{'message': 'A'}, {'message': 'B'}]
     repo = backend.create_repo(commits=commits)
     scm.ScmModel().mark_for_invalidation(repo.repo_name, delete=True)
     assert repo.changeset_cache['revision'] == 1
 
 
 def test_mark_for_invalidation_with_delete_updates_last_commit_empty(backend):
     repo = backend.create_repo()
     scm.ScmModel().mark_for_invalidation(repo.repo_name, delete=True)
     assert repo.changeset_cache['revision'] == -1
 
 
 def test_strip_with_multiple_heads(backend_hg):
     commits = [
         {'message': 'A'},
         {'message': 'a'},
         {'message': 'b'},
-        {'message': 'B', 'parents': ['A']},
-        {'message': 'a1'},
+        {'message': 'B', 'parents': ['A'], 'branch': 'feature'},
+        {'message': 'a1', 'branch': 'feature'},
     ]
     repo = backend_hg.create_repo(commits=commits)
     commit_ids = backend_hg.commit_ids
 
     model = scm.ScmModel()
     model.strip(repo, commit_ids['b'], branch=None)
 
     vcs_repo = repo.scm_instance()
     rest_commit_ids = [c.raw_id for c in vcs_repo.get_commits()]
     assert len(rest_commit_ids) == 4
     assert commit_ids['b'] not in rest_commit_ids
 
 
 def test_strip_with_single_heads(backend_hg):
     commits = [
         {'message': 'A'},
         {'message': 'a'},
         {'message': 'b'},
     ]
     repo = backend_hg.create_repo(commits=commits)
     commit_ids = backend_hg.commit_ids
 
     model = scm.ScmModel()
     model.strip(repo, commit_ids['b'], branch=None)
 
     vcs_repo = repo.scm_instance()
     rest_commit_ids = [c.raw_id for c in vcs_repo.get_commits()]
     assert len(rest_commit_ids) == 2
     assert commit_ids['b'] not in rest_commit_ids
 
 
 def test_get_nodes_returns_unicode_flat(backend):
     repo = backend.repo
     commit_id = repo.get_commit(commit_idx=0).raw_id
     directories, files = scm.ScmModel().get_nodes(repo.repo_name, commit_id, flat=True)
     assert_contains_only_str_chars(directories)
     assert_contains_only_str_chars(files)
 
 
 def test_get_nodes_returns_unicode_non_flat(backend):
     repo = backend.repo
     commit_id = repo.get_commit(commit_idx=0).raw_id
 
     directories, files = scm.ScmModel().get_nodes(repo.repo_name, commit_id, flat=False)
     # johbo: Checking only the names for now, since that is the critical
     # part.
     assert_contains_only_str_chars([d['name'] for d in directories])
     assert_contains_only_str_chars([f['name'] for f in files])
 
 
 def test_get_nodes_max_file_bytes(backend_random):
     repo = backend_random.repo
     max_file_bytes = 10
     directories, files = scm.ScmModel().get_nodes(
         repo.repo_name, repo.get_commit(commit_idx=0).raw_id, content=True,
         extended_info=True, flat=False)
     assert any(file['content'] and len(file['content']) > max_file_bytes
                for file in files)
 
     directories, files = scm.ScmModel().get_nodes(
         repo.repo_name, repo.get_commit(commit_idx=0).raw_id, content=True,
         extended_info=True, flat=False, max_file_bytes=max_file_bytes)
     assert all(
         file['content'] is None if file['size'] > max_file_bytes else True
         for file in files)
 
 
 def assert_contains_only_str_chars(structure):
     assert structure
     for value in structure:
         assert isinstance(value, str)
 
 
 @pytest.mark.backends("hg", "git")
 def test_get_non_str_reference(backend):
     model = scm.ScmModel()
     special_name = "Adını"
     non_str_list = [special_name]
 
     def scm_instance():
         return Mock(
             branches=non_str_list, bookmarks=non_str_list,
             tags=non_str_list, alias=backend.alias)
 
     repo = Mock(__class__=db.Repository, scm_instance=scm_instance)
     choices, __ = model.get_repo_landing_revs(translator=lambda s: s, repo=repo)
     if backend.alias == 'hg':
         valid_choices = [
             'rev:tip', f'branch:{special_name}',
             f'book:{special_name}', f'tag:{special_name}']
     else:
         valid_choices = [
             'rev:tip', f'branch:{special_name}',
             f'tag:{special_name}']
 
     assert choices == valid_choices

rhodecode/tests/utils.py

@@ -1,493 +1,494 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 threading
 import time
 import sys
 import logging
 import os.path
 import subprocess
 import tempfile
 import urllib.request
 import urllib.error
 import urllib.parse
 from lxml.html import fromstring, tostring
 from lxml.cssselect import CSSSelector
 from urllib.parse import unquote_plus
 import webob
 
 from webtest.app import TestResponse, TestApp
 
 
 import pytest
 
 try:
     import rc_testdata
 except ImportError:
     raise ImportError('Failed to import rc_testdata, '
                       'please make sure this package is installed from requirements_test.txt')
 
 from rhodecode.model.db import User, Repository
 from rhodecode.model.meta import Session
 from rhodecode.model.scm import ScmModel
 from rhodecode.lib.vcs.backends.svn.repository import SubversionRepository
 from rhodecode.lib.vcs.backends.base import EmptyCommit
 from rhodecode.tests import login_user_session
 
 log = logging.getLogger(__name__)
 
 
 def print_to_func(value, print_to=sys.stderr):
     print(value, file=print_to)
 
 
 class CustomTestResponse(TestResponse):
 
     def _save_output(self, out):
         f = tempfile.NamedTemporaryFile(mode='w', delete=False, prefix='rc-test-', suffix='.html')
         f.write(out)
         return f.name
 
     def mustcontain(self, *strings, **kw):
         """
         Assert that the response contains all the strings passed
         in as arguments.
 
         Equivalent to::
 
             assert string in res
         """
         print_body = kw.pop('print_body', False)
         print_to = kw.pop('print_to', sys.stderr)
 
         if 'no' in kw:
             no = kw['no']
             del kw['no']
             if isinstance(no, str):
                 no = [no]
         else:
             no = []
         if kw:
             raise TypeError(f"The only keyword argument allowed is 'no' got {kw}")
 
         f = self._save_output(str(self))
 
         for s in strings:
             if s not in self:
                 print_to_func(f"Actual response (no {s!r}):", print_to=print_to)
                 print_to_func(f"body output saved as `{f}`", print_to=print_to)
                 if print_body:
                     print_to_func(str(self), print_to=print_to)
                 raise IndexError(f"Body does not contain string {s!r}, body output saved as {f}")
 
         for no_s in no:
             if no_s in self:
                 print_to_func(f"Actual response (has {no_s!r})", print_to=print_to)
                 print_to_func(f"body output saved as `{f}`", print_to=print_to)
                 if print_body:
                     print_to_func(str(self), print_to=print_to)
                 raise IndexError(f"Body contains bad string {no_s!r}, body output saved as {f}")
 
     def assert_response(self):
         return AssertResponse(self)
 
     def get_session_from_response(self):
         """
         This returns the session from a response object.
         """
         from rhodecode.lib.rc_beaker import session_factory_from_settings
         session = session_factory_from_settings(self.test_app._pyramid_settings)
         return session(self.request)
 
 
 class TestRequest(webob.BaseRequest):
 
     # for py.test, so it doesn't try to run this tas by name starting with test...
     disabled = True
     ResponseClass = CustomTestResponse
 
     def add_response_callback(self, callback):
         pass
 
     @classmethod
     def blank(cls, path, environ=None, base_url=None,
               headers=None, POST=None, **kw):
 
         if not path.isascii():
             # our custom quote path if it contains non-ascii chars
             path = urllib.parse.quote(path)
 
         return super(TestRequest, cls).blank(
             path, environ=environ, base_url=base_url, headers=headers, POST=POST, **kw)
 
 
 class CustomTestApp(TestApp):
     """
     Custom app to make mustcontain more Useful, and extract special methods
     """
     RequestClass = TestRequest
     rc_login_data = {}
     rc_current_session = None
 
     def login(self, username=None, password=None):
         from rhodecode.lib import auth
 
         if username and password:
             session = login_user_session(self, username, password)
         else:
             session = login_user_session(self)
 
         self.rc_login_data['csrf_token'] = auth.get_csrf_token(session)
         self.rc_current_session = session
         return session['rhodecode_user']
 
     @property
     def csrf_token(self):
         return self.rc_login_data['csrf_token']
 
     @property
     def _pyramid_registry(self):
         return self.app.config.registry
 
     @property
     def _pyramid_settings(self):
         return self._pyramid_registry.settings
 
     def do_request(self, req, status=None, expect_errors=None):
         # you can put custom code here
         return super().do_request(req, status, expect_errors)
 
 
 def set_anonymous_access(enabled):
     """(Dis)allows anonymous access depending on parameter `enabled`"""
     user = User.get_default_user()
     user.active = enabled
     Session().add(user)
     Session().commit()
     time.sleep(1.5)  # must sleep for cache (1s to expire)
     log.info('anonymous access is now: %s', enabled)
     assert enabled == User.get_default_user().active, (
         'Cannot set anonymous access')
 
 
 def check_xfail_backends(node, backend_alias):
     # Using "xfail_backends" here intentionally, since this marks work
     # which is "to be done" soon.
     skip_marker = node.get_closest_marker('xfail_backends')
     if skip_marker and backend_alias in skip_marker.args:
         msg = "Support for backend %s to be developed." % (backend_alias, )
         msg = skip_marker.kwargs.get('reason', msg)
         pytest.xfail(msg)
 
 
 def check_skip_backends(node, backend_alias):
     # Using "skip_backends" here intentionally, since this marks work which is
     # not supported.
     skip_marker = node.get_closest_marker('skip_backends')
     if skip_marker and backend_alias in skip_marker.args:
         msg = "Feature not supported for backend %s." % (backend_alias, )
         msg = skip_marker.kwargs.get('reason', msg)
         pytest.skip(msg)
 
 
 def extract_git_repo_from_dump(dump_name, repo_name):
     """Create git repo `repo_name` from dump `dump_name`."""
     repos_path = ScmModel().repos_path
     target_path = os.path.join(repos_path, repo_name)
     rc_testdata.extract_git_dump(dump_name, target_path)
     return target_path
 
 
 def extract_hg_repo_from_dump(dump_name, repo_name):
     """Create hg repo `repo_name` from dump `dump_name`."""
     repos_path = ScmModel().repos_path
     target_path = os.path.join(repos_path, repo_name)
     rc_testdata.extract_hg_dump(dump_name, target_path)
     return target_path
 
 
 def extract_svn_repo_from_dump(dump_name, repo_name):
     """Create a svn repo `repo_name` from dump `dump_name`."""
     repos_path = ScmModel().repos_path
     target_path = os.path.join(repos_path, repo_name)
     SubversionRepository(target_path, create=True)
     _load_svn_dump_into_repo(dump_name, target_path)
     return target_path
 
 
 def assert_message_in_log(log_records, message, levelno, module):
     messages = [
         r.message for r in log_records
         if r.module == module and r.levelno == levelno
     ]
     assert message in messages
 
 
 def _load_svn_dump_into_repo(dump_name, repo_path):
     """
     Utility to populate a svn repository with a named dump
 
     Currently the dumps are in rc_testdata. They might later on be
     integrated with the main repository once they stabilize more.
     """
     dump = rc_testdata.load_svn_dump(dump_name)
     load_dump = subprocess.Popen(
         ['svnadmin', 'load', repo_path],
         stdin=subprocess.PIPE, stdout=subprocess.PIPE,
         stderr=subprocess.PIPE)
     out, err = load_dump.communicate(dump)
     if load_dump.returncode != 0:
         log.error("Output of load_dump command: %s", out)
         log.error("Error output of load_dump command: %s", err)
         raise Exception(
             'Failed to load dump "%s" into repository at path "%s".'
             % (dump_name, repo_path))
 
 
 class AssertResponse(object):
     """
     Utility that helps to assert things about a given HTML response.
     """
 
     def __init__(self, response):
         self.response = response
 
     def get_imports(self):
         return fromstring, tostring, CSSSelector
 
     def one_element_exists(self, css_selector):
         self.get_element(css_selector)
 
     def no_element_exists(self, css_selector):
         assert not self._get_elements(css_selector)
 
     def element_equals_to(self, css_selector, expected_content):
         element = self.get_element(css_selector)
         element_text = self._element_to_string(element)
 
         assert expected_content in element_text
 
     def element_contains(self, css_selector, expected_content):
         element = self.get_element(css_selector)
         assert expected_content in element.text_content()
 
     def element_value_contains(self, css_selector, expected_content):
         element = self.get_element(css_selector)
         assert expected_content in element.value
 
     def contains_one_link(self, link_text, href):
         fromstring, tostring, CSSSelector = self.get_imports()
         doc = fromstring(self.response.body)
         sel = CSSSelector('a[href]')
         elements = [
             e for e in sel(doc) if e.text_content().strip() == link_text]
         assert len(elements) == 1, "Did not find link or found multiple links"
         self._ensure_url_equal(elements[0].attrib.get('href'), href)
 
     def contains_one_anchor(self, anchor_id):
         fromstring, tostring, CSSSelector = self.get_imports()
         doc = fromstring(self.response.body)
         sel = CSSSelector('#' + anchor_id)
         elements = sel(doc)
         assert len(elements) == 1, 'cannot find 1 element {}'.format(anchor_id)
 
     def _ensure_url_equal(self, found, expected):
         assert _Url(found) == _Url(expected)
 
     def get_element(self, css_selector):
         elements = self._get_elements(css_selector)
         assert len(elements) == 1, 'cannot find 1 element {}'.format(css_selector)
         return elements[0]
 
     def get_elements(self, css_selector):
         return self._get_elements(css_selector)
 
     def _get_elements(self, css_selector):
         fromstring, tostring, CSSSelector = self.get_imports()
         doc = fromstring(self.response.body)
         sel = CSSSelector(css_selector)
         elements = sel(doc)
         return elements
 
     def _element_to_string(self, element):
         fromstring, tostring, CSSSelector = self.get_imports()
         return tostring(element, encoding='unicode')
 
 
 class _Url(object):
     """
     A url object that can be compared with other url orbjects
     without regard to the vagaries of encoding, escaping, and ordering
     of parameters in query strings.
 
     Inspired by
     http://stackoverflow.com/questions/5371992/comparing-two-urls-in-python
     """
 
     def __init__(self, url):
         parts = urllib.parse.urlparse(url)
         _query = frozenset(urllib.parse.parse_qsl(parts.query))
         _path = unquote_plus(parts.path)
         parts = parts._replace(query=_query, path=_path)
         self.parts = parts
 
     def __eq__(self, other):
         return self.parts == other.parts
 
     def __hash__(self):
         return hash(self.parts)
 
 
 def run_test_concurrently(times, raise_catched_exc=True):
     """
     Add this decorator to small pieces of code that you want to test
     concurrently
 
     ex:
 
     @test_concurrently(25)
     def my_test_function():
         ...
     """
     def test_concurrently_decorator(test_func):
         def wrapper(*args, **kwargs):
             exceptions = []
 
             def call_test_func():
                 try:
                     test_func(*args, **kwargs)
                 except Exception as e:
                     exceptions.append(e)
                     if raise_catched_exc:
                         raise
             threads = []
             for i in range(times):
                 threads.append(threading.Thread(target=call_test_func))
             for t in threads:
                 t.start()
             for t in threads:
                 t.join()
             if exceptions:
                 raise Exception(
                     'test_concurrently intercepted %s exceptions: %s' % (
                         len(exceptions), exceptions))
         return wrapper
     return test_concurrently_decorator
 
 
 def wait_for_url(url, timeout=10):
     """
     Wait until URL becomes reachable.
 
     It polls the URL until the timeout is reached or it became reachable.
     If will call to `py.test.fail` in case the URL is not reachable.
     """
     timeout = time.time() + timeout
     last = 0
     wait = 0.1
 
     while timeout > last:
         last = time.time()
         if is_url_reachable(url, log_exc=False):
             break
         elif (last + wait) > time.time():
             # Go to sleep because not enough time has passed since last check.
             time.sleep(wait)
     else:
         pytest.fail(f"Timeout while waiting for URL {url}")
 
 
 def is_url_reachable(url: str, log_exc: bool = False) -> bool:
     try:
         urllib.request.urlopen(url)
     except urllib.error.URLError:
         if log_exc:
             log.exception(f'URL `{url}` reach error')
         return False
     return True
 
 
 def repo_on_filesystem(repo_name):
     from rhodecode.lib import vcs
     from rhodecode.tests import TESTS_TMP_PATH
     repo = vcs.get_vcs_instance(
         os.path.join(TESTS_TMP_PATH, repo_name), create=False)
     return repo is not None
 
 
 def commit_change(
-        repo, filename: bytes, content: bytes, message, vcs_type, parent=None, newfile=False):
+        repo, filename: bytes, content: bytes, message, vcs_type, parent=None, branch=None, newfile=False):
     from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 
     repo = Repository.get_by_repo_name(repo)
     _commit = parent
     if not parent:
         _commit = EmptyCommit(alias=vcs_type)
 
     if newfile:
         nodes = {
             filename: {
                 'content': content
             }
         }
         commit = ScmModel().create_nodes(
             user=TEST_USER_ADMIN_LOGIN, repo=repo,
             message=message,
             nodes=nodes,
             parent_commit=_commit,
             author=f'{TEST_USER_ADMIN_LOGIN} <[e-mail]>',
         )
     else:
         commit = ScmModel().commit_change(
             repo=repo.scm_instance(), repo_name=repo.repo_name,
             commit=parent, user=TEST_USER_ADMIN_LOGIN,
             author=f'{TEST_USER_ADMIN_LOGIN} <[e-mail]>',
             message=message,
             content=content,
-            f_path=filename
+            f_path=filename,
+            branch=branch
         )
     return commit
 
 
 def permission_update_data_generator(csrf_token, default=None, grant=None, revoke=None):
     if not default:
         raise ValueError('Permission for default user must be given')
     form_data = [(
         'csrf_token', csrf_token
     )]
     # add default
     form_data.extend([
         ('u_perm_1', default)
     ])
 
     if grant:
         for cnt, (obj_id, perm, obj_name, obj_type) in enumerate(grant, 1):
             form_data.extend([
                 ('perm_new_member_perm_new{}'.format(cnt), perm),
                 ('perm_new_member_id_new{}'.format(cnt), obj_id),
                 ('perm_new_member_name_new{}'.format(cnt), obj_name),
                 ('perm_new_member_type_new{}'.format(cnt), obj_type),
 
             ])
     if revoke:
         for obj_id, obj_type in revoke:
             form_data.extend([
                 ('perm_del_member_id_{}'.format(obj_id), obj_id),
                 ('perm_del_member_type_{}'.format(obj_id), obj_type),
             ])
     return form_data