pull-requests: validate ref types for pull request so users cannot provide wrongs ones.

rhodecode/api/tests/test_create_pull_request.py

@@ -1,349 +1,368 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2018 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'
         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']},
         ]
         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/api/tests/test_utils.py

@@ -1,295 +1,295 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2018 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
 
 from rhodecode.api import utils
 from rhodecode.api import JSONRPCError
 from rhodecode.lib.vcs.exceptions import RepositoryError
 
 
 class TestGetCommitOrError(object):
     def setup(self):
         self.commit_hash = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa10'
 
     @pytest.mark.parametrize("ref", ['ref', '12345', 'a:b:c:d', 'branch:name'])
     def test_ref_cannot_be_parsed(self, ref):
         repo = Mock()
         with pytest.raises(JSONRPCError) as excinfo:
             utils.get_commit_or_error(ref, repo)
         expected_message = (
             'Ref `{ref}` given in a wrong format. Please check the API'
             ' documentation for more details'.format(ref=ref)
         )
         assert excinfo.value.message == expected_message
 
     def test_success_with_hash_specified(self):
         repo = Mock()
         ref_type = 'branch'
         ref = '{}:master:{}'.format(ref_type, self.commit_hash)
 
         with patch('rhodecode.api.utils.get_commit_from_ref_name') as get_commit:
             result = utils.get_commit_or_error(ref, repo)
             get_commit.assert_called_once_with(
                 repo, self.commit_hash)
             assert result == get_commit()
 
     def test_raises_an_error_when_commit_not_found(self):
         repo = Mock()
         ref = 'branch:master:{}'.format(self.commit_hash)
 
         with patch('rhodecode.api.utils.get_commit_from_ref_name') as get_commit:
             get_commit.side_effect = RepositoryError('Commit not found')
             with pytest.raises(JSONRPCError) as excinfo:
                 utils.get_commit_or_error(ref, repo)
             expected_message = 'Ref `{}` does not exist'.format(ref)
             assert excinfo.value.message == expected_message
 
 
 class TestResolveRefOrError(object):
     def setup(self):
         self.commit_hash = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa10'
 
     def test_success_with_no_hash_specified(self):
         repo = Mock()
         ref_type = 'branch'
         ref_name = 'master'
         ref = '{}:{}'.format(ref_type, ref_name)
 
         with patch('rhodecode.api.utils._get_ref_hash') \
                 as _get_ref_hash:
             _get_ref_hash.return_value = self.commit_hash
             result = utils.resolve_ref_or_error(ref, repo)
             _get_ref_hash.assert_called_once_with(repo, ref_type, ref_name)
             assert result == '{}:{}'.format(ref, self.commit_hash)
 
     def test_non_supported_refs(self):
         repo = Mock()
-        ref = 'ancestor:ref'
+        ref = 'bookmark:ref'
         with pytest.raises(JSONRPCError) as excinfo:
             utils.resolve_ref_or_error(ref, repo)
         expected_message = (
-            'The specified value:ancestor:`ref` does not exist, or is not allowed.')
+            'The specified value:bookmark:`ref` does not exist, or is not allowed.')
         assert excinfo.value.message == expected_message
 
     def test_branch_is_not_found(self):
         repo = Mock()
         ref = 'branch:non-existing-one'
         with patch('rhodecode.api.utils._get_ref_hash')\
                 as _get_ref_hash:
             _get_ref_hash.side_effect = KeyError()
             with pytest.raises(JSONRPCError) as excinfo:
                 utils.resolve_ref_or_error(ref, repo)
         expected_message = (
             'The specified value:branch:`non-existing-one` does not exist, or is not allowed.')
         assert excinfo.value.message == expected_message
 
     def test_bookmark_is_not_found(self):
         repo = Mock()
         ref = 'bookmark:non-existing-one'
         with patch('rhodecode.api.utils._get_ref_hash')\
                 as _get_ref_hash:
             _get_ref_hash.side_effect = KeyError()
             with pytest.raises(JSONRPCError) as excinfo:
                 utils.resolve_ref_or_error(ref, repo)
         expected_message = (
             'The specified value:bookmark:`non-existing-one` does not exist, or is not allowed.')
         assert excinfo.value.message == expected_message
 
     @pytest.mark.parametrize("ref", ['ref', '12345', 'a:b:c:d'])
     def test_ref_cannot_be_parsed(self, ref):
         repo = Mock()
         with pytest.raises(JSONRPCError) as excinfo:
             utils.resolve_ref_or_error(ref, repo)
         expected_message = (
             'Ref `{ref}` given in a wrong format. Please check the API'
             ' documentation for more details'.format(ref=ref)
         )
         assert excinfo.value.message == expected_message
 
 
 class TestGetRefHash(object):
     def setup(self):
         self.commit_hash = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa10'
         self.bookmark_name = 'test-bookmark'
 
     @pytest.mark.parametrize("alias, branch_name", [
         ("git", "master"),
         ("hg", "default")
     ])
     def test_returns_hash_by_branch_name(self, alias, branch_name):
         with patch('rhodecode.model.db.Repository') as repo:
             repo.scm_instance().alias = alias
             repo.scm_instance().branches = {branch_name: self.commit_hash}
             result_hash = utils._get_ref_hash(repo, 'branch', branch_name)
             assert result_hash == self.commit_hash
 
     @pytest.mark.parametrize("alias, branch_name", [
         ("git", "master"),
         ("hg", "default")
     ])
     def test_raises_error_when_branch_is_not_found(self, alias, branch_name):
         with patch('rhodecode.model.db.Repository') as repo:
             repo.scm_instance().alias = alias
             repo.scm_instance().branches = {}
             with pytest.raises(KeyError):
                 utils._get_ref_hash(repo, 'branch', branch_name)
 
     def test_returns_hash_when_bookmark_is_specified_for_hg(self):
         with patch('rhodecode.model.db.Repository') as repo:
             repo.scm_instance().alias = 'hg'
             repo.scm_instance().bookmarks = {
                 self.bookmark_name: self.commit_hash}
             result_hash = utils._get_ref_hash(
                 repo, 'bookmark', self.bookmark_name)
             assert result_hash == self.commit_hash
 
     def test_raises_error_when_bookmark_is_not_found_in_hg_repo(self):
         with patch('rhodecode.model.db.Repository') as repo:
             repo.scm_instance().alias = 'hg'
             repo.scm_instance().bookmarks = {}
             with pytest.raises(KeyError):
                 utils._get_ref_hash(repo, 'bookmark', self.bookmark_name)
 
     def test_raises_error_when_bookmark_is_specified_for_git(self):
         with patch('rhodecode.model.db.Repository') as repo:
             repo.scm_instance().alias = 'git'
             repo.scm_instance().bookmarks = {
                 self.bookmark_name: self.commit_hash}
             with pytest.raises(ValueError):
                 utils._get_ref_hash(repo, 'bookmark', self.bookmark_name)
 
 
 class TestUserByNameOrError(object):
     def test_user_found_by_id(self):
         fake_user = Mock(id=123)
 
         patcher = patch('rhodecode.model.user.UserModel.get_user')
         with patcher as get_user:
             get_user.return_value = fake_user
 
             patcher = patch('rhodecode.model.user.UserModel.get_by_username')
             with patcher as get_by_username:
                 result = utils.get_user_or_error(123)
                 assert result == fake_user
 
     def test_user_not_found_by_id_as_str(self):
         fake_user = Mock(id=123)
 
         patcher = patch('rhodecode.model.user.UserModel.get_user')
         with patcher as get_user:
             get_user.return_value = fake_user
             patcher = patch('rhodecode.model.user.UserModel.get_by_username')
             with patcher as get_by_username:
                 get_by_username.return_value = None
 
                 with pytest.raises(JSONRPCError):
                     utils.get_user_or_error('123')
 
     def test_user_found_by_name(self):
         fake_user = Mock(id=123)
 
         patcher = patch('rhodecode.model.user.UserModel.get_user')
         with patcher as get_user:
             get_user.return_value = None
 
             patcher = patch('rhodecode.model.user.UserModel.get_by_username')
             with patcher as get_by_username:
                 get_by_username.return_value = fake_user
 
                 result = utils.get_user_or_error('test')
                 assert result == fake_user
 
     def test_user_not_found_by_id(self):
         patcher = patch('rhodecode.model.user.UserModel.get_user')
         with patcher as get_user:
             get_user.return_value = None
             patcher = patch('rhodecode.model.user.UserModel.get_by_username')
             with patcher as get_by_username:
                 get_by_username.return_value = None
 
                 with pytest.raises(JSONRPCError) as excinfo:
                     utils.get_user_or_error(123)
 
                 expected_message = 'user `123` does not exist'
                 assert excinfo.value.message == expected_message
 
     def test_user_not_found_by_name(self):
         patcher = patch('rhodecode.model.user.UserModel.get_by_username')
         with patcher as get_by_username:
             get_by_username.return_value = None
             with pytest.raises(JSONRPCError) as excinfo:
                 utils.get_user_or_error('test')
 
         expected_message = 'user `test` does not exist'
         assert excinfo.value.message == expected_message
 
 
 class TestGetCommitDict(object):
     @pytest.mark.parametrize('filename, expected', [
         (b'sp\xc3\xa4cial', u'sp\xe4cial'),
         (b'sp\xa4cial', u'sp\ufffdcial'),
     ])
     def test_decodes_filenames_to_unicode(self, filename, expected):
         result = utils._get_commit_dict(filename=filename, op='A')
         assert result['filename'] == expected
 
 
 class TestRepoAccess(object):
     def setup_method(self, method):
 
         self.admin_perm_patch = patch(
             'rhodecode.api.utils.HasPermissionAnyApi')
         self.repo_perm_patch = patch(
             'rhodecode.api.utils.HasRepoPermissionAnyApi')
 
     def test_has_superadmin_permission_checks_for_admin(self):
         admin_mock = Mock()
         with self.admin_perm_patch as amock:
             amock.return_value = admin_mock
             assert utils.has_superadmin_permission('fake_user')
             amock.assert_called_once_with('hg.admin')
 
         admin_mock.assert_called_once_with(user='fake_user')
 
     def test_has_repo_permissions_checks_for_repo_access(self):
         repo_mock = Mock()
         fake_repo = Mock()
         with self.repo_perm_patch as rmock:
             rmock.return_value = repo_mock
             assert utils.validate_repo_permissions(
                 'fake_user', 'fake_repo_id', fake_repo,
                 ['perm1', 'perm2'])
             rmock.assert_called_once_with(*['perm1', 'perm2'])
 
         repo_mock.assert_called_once_with(
             user='fake_user', repo_name=fake_repo.repo_name)
 
     def test_has_repo_permissions_raises_not_found(self):
         repo_mock = Mock(return_value=False)
         fake_repo = Mock()
         with self.repo_perm_patch as rmock:
             rmock.return_value = repo_mock
             with pytest.raises(JSONRPCError) as excinfo:
                 utils.validate_repo_permissions(
                     'fake_user', 'fake_repo_id', fake_repo, 'perms')
                 assert 'fake_repo_id' in excinfo

rhodecode/api/utils.py

@@ -1,441 +1,449 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2014-2018 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/
 
 """
 JSON RPC utils
 """
 
 import collections
 import logging
 
 from rhodecode.api.exc import JSONRPCError
 from rhodecode.lib.auth import (
     HasPermissionAnyApi, HasRepoPermissionAnyApi, HasRepoGroupPermissionAnyApi)
 from rhodecode.lib.utils import safe_unicode
 from rhodecode.lib.vcs.exceptions import RepositoryError
 from rhodecode.controllers.utils import get_commit_from_ref_name
 from rhodecode.lib.utils2 import str2bool
 
 log = logging.getLogger(__name__)
 
 
 class OAttr(object):
     """
     Special Option that defines other attribute, and can default to them
 
     Example::
 
         def test(apiuser, userid=Optional(OAttr('apiuser')):
             user = Optional.extract(userid, evaluate_locals=local())
             #if we pass in userid, we get it, else it will default to apiuser
             #attribute
     """
 
     def __init__(self, attr_name):
         self.attr_name = attr_name
 
     def __repr__(self):
         return '<OptionalAttr:%s>' % self.attr_name
 
     def __call__(self):
         return self
 
 
 class Optional(object):
     """
     Defines an optional parameter::
 
         param = param.getval() if isinstance(param, Optional) else param
         param = param() if isinstance(param, Optional) else param
 
     is equivalent of::
 
         param = Optional.extract(param)
 
     """
 
     def __init__(self, type_):
         self.type_ = type_
 
     def __repr__(self):
         return '<Optional:%s>' % self.type_.__repr__()
 
     def __call__(self):
         return self.getval()
 
     def getval(self, evaluate_locals=None):
         """
         returns value from this Optional instance
         """
         if isinstance(self.type_, OAttr):
             param_name = self.type_.attr_name
             if evaluate_locals:
                 return evaluate_locals[param_name]
             # use params name
             return param_name
         return self.type_
 
     @classmethod
     def extract(cls, val, evaluate_locals=None, binary=None):
         """
         Extracts value from Optional() instance
 
         :param val:
         :return: original value if it's not Optional instance else
             value of instance
         """
         if isinstance(val, cls):
             val = val.getval(evaluate_locals)
 
         if binary:
             val = str2bool(val)
 
         return val
 
 
 def parse_args(cli_args, key_prefix=''):
     from rhodecode.lib.utils2 import (escape_split)
     kwargs = collections.defaultdict(dict)
     for el in escape_split(cli_args, ','):
         kv = escape_split(el, '=', 1)
         if len(kv) == 2:
             k, v = kv
             kwargs[key_prefix + k] = v
     return kwargs
 
 
 def get_origin(obj):
     """
     Get origin of permission from object.
 
     :param obj:
     """
     origin = 'permission'
 
     if getattr(obj, 'owner_row', '') and getattr(obj, 'admin_row', ''):
         # admin and owner case, maybe we should use dual string ?
         origin = 'owner'
     elif getattr(obj, 'owner_row', ''):
         origin = 'owner'
     elif getattr(obj, 'admin_row', ''):
         origin = 'super-admin'
     return origin
 
 
 def store_update(updates, attr, name):
     """
     Stores param in updates dict if it's not instance of Optional
     allows easy updates of passed in params
     """
     if not isinstance(attr, Optional):
         updates[name] = attr
 
 
 def has_superadmin_permission(apiuser):
     """
     Return True if apiuser is admin or return False
 
     :param apiuser:
     """
     if HasPermissionAnyApi('hg.admin')(user=apiuser):
         return True
     return False
 
 
 def validate_repo_permissions(apiuser, repoid, repo, perms):
     """
     Raise JsonRPCError if apiuser is not authorized or return True
 
     :param apiuser:
     :param repoid:
     :param repo:
     :param perms:
     """
     if not HasRepoPermissionAnyApi(*perms)(
             user=apiuser, repo_name=repo.repo_name):
         raise JSONRPCError(
             'repository `%s` does not exist' % repoid)
 
     return True
 
 
 def validate_repo_group_permissions(apiuser, repogroupid, repo_group, perms):
     """
     Raise JsonRPCError if apiuser is not authorized or return True
 
     :param apiuser:
     :param repogroupid: just the id of repository group
     :param repo_group: instance of repo_group
     :param perms:
     """
     if not HasRepoGroupPermissionAnyApi(*perms)(
             user=apiuser, group_name=repo_group.group_name):
         raise JSONRPCError(
             'repository group `%s` does not exist' % repogroupid)
 
     return True
 
 
 def validate_set_owner_permissions(apiuser, owner):
     if isinstance(owner, Optional):
         owner = get_user_or_error(apiuser.user_id)
     else:
         if has_superadmin_permission(apiuser):
             owner = get_user_or_error(owner)
         else:
             # forbid setting owner for non-admins
             raise JSONRPCError(
                 'Only RhodeCode super-admin can specify `owner` param')
     return owner
 
 
 def get_user_or_error(userid):
     """
     Get user by id or name or return JsonRPCError if not found
 
     :param userid:
     """
     from rhodecode.model.user import UserModel
     user_model = UserModel()
 
     if isinstance(userid, (int, long)):
         try:
             user = user_model.get_user(userid)
         except ValueError:
             user = None
     else:
         user = user_model.get_by_username(userid)
 
     if user is None:
         raise JSONRPCError(
             'user `%s` does not exist' % (userid,))
     return user
 
 
 def get_repo_or_error(repoid):
     """
     Get repo by id or name or return JsonRPCError if not found
 
     :param repoid:
     """
     from rhodecode.model.repo import RepoModel
     repo_model = RepoModel()
 
     if isinstance(repoid, (int, long)):
         try:
             repo = repo_model.get_repo(repoid)
         except ValueError:
             repo = None
     else:
         repo = repo_model.get_by_repo_name(repoid)
 
     if repo is None:
         raise JSONRPCError(
             'repository `%s` does not exist' % (repoid,))
     return repo
 
 
 def get_repo_group_or_error(repogroupid):
     """
     Get repo group by id or name or return JsonRPCError if not found
 
     :param repogroupid:
     """
     from rhodecode.model.repo_group import RepoGroupModel
     repo_group_model = RepoGroupModel()
 
     if isinstance(repogroupid, (int, long)):
         try:
             repo_group = repo_group_model._get_repo_group(repogroupid)
         except ValueError:
             repo_group = None
     else:
         repo_group = repo_group_model.get_by_group_name(repogroupid)
 
     if repo_group is None:
         raise JSONRPCError(
             'repository group `%s` does not exist' % (repogroupid,))
     return repo_group
 
 
 def get_user_group_or_error(usergroupid):
     """
     Get user group by id or name or return JsonRPCError if not found
 
     :param usergroupid:
     """
     from rhodecode.model.user_group import UserGroupModel
     user_group_model = UserGroupModel()
 
     if isinstance(usergroupid, (int, long)):
         try:
             user_group = user_group_model.get_group(usergroupid)
         except ValueError:
             user_group = None
     else:
         user_group = user_group_model.get_by_name(usergroupid)
 
     if user_group is None:
         raise JSONRPCError(
             'user group `%s` does not exist' % (usergroupid,))
     return user_group
 
 
 def get_perm_or_error(permid, prefix=None):
     """
     Get permission by id or name or return JsonRPCError if not found
 
     :param permid:
     """
     from rhodecode.model.permission import PermissionModel
 
     perm = PermissionModel.cls.get_by_key(permid)
     if perm is None:
         raise JSONRPCError('permission `%s` does not exist' % (permid,))
     if prefix:
         if not perm.permission_name.startswith(prefix):
             raise JSONRPCError('permission `%s` is invalid, '
                                'should start with %s' % (permid, prefix))
     return perm
 
 
 def get_gist_or_error(gistid):
     """
     Get gist by id or gist_access_id or return JsonRPCError if not found
 
     :param gistid:
     """
     from rhodecode.model.gist import GistModel
 
     gist = GistModel.cls.get_by_access_id(gistid)
     if gist is None:
         raise JSONRPCError('gist `%s` does not exist' % (gistid,))
     return gist
 
 
 def get_pull_request_or_error(pullrequestid):
     """
     Get pull request by id or return JsonRPCError if not found
 
     :param pullrequestid:
     """
     from rhodecode.model.pull_request import PullRequestModel
 
     try:
         pull_request = PullRequestModel().get(int(pullrequestid))
     except ValueError:
         raise JSONRPCError('pullrequestid must be an integer')
     if not pull_request:
         raise JSONRPCError('pull request `%s` does not exist' % (
             pullrequestid,))
     return pull_request
 
 
 def build_commit_data(commit, detail_level):
     parsed_diff = []
     if detail_level == 'extended':
         for f in commit.added:
             parsed_diff.append(_get_commit_dict(filename=f.path, op='A'))
         for f in commit.changed:
             parsed_diff.append(_get_commit_dict(filename=f.path, op='M'))
         for f in commit.removed:
             parsed_diff.append(_get_commit_dict(filename=f.path, op='D'))
 
     elif detail_level == 'full':
         from rhodecode.lib.diffs import DiffProcessor
         diff_processor = DiffProcessor(commit.diff())
         for dp in diff_processor.prepare():
             del dp['stats']['ops']
             _stats = dp['stats']
             parsed_diff.append(_get_commit_dict(
                 filename=dp['filename'], op=dp['operation'],
                 new_revision=dp['new_revision'],
                 old_revision=dp['old_revision'],
                 raw_diff=dp['raw_diff'], stats=_stats))
 
     return parsed_diff
 
 
 def get_commit_or_error(ref, repo):
     try:
         ref_type, _, ref_hash = ref.split(':')
     except ValueError:
         raise JSONRPCError(
             'Ref `{ref}` given in a wrong format. Please check the API'
             ' documentation for more details'.format(ref=ref))
     try:
         # TODO: dan: refactor this to use repo.scm_instance().get_commit()
         # once get_commit supports ref_types
         return get_commit_from_ref_name(repo, ref_hash)
     except RepositoryError:
         raise JSONRPCError('Ref `{ref}` does not exist'.format(ref=ref))
 
 
-def resolve_ref_or_error(ref, repo):
+def _get_ref_hash(repo, type_, name):
+    vcs_repo = repo.scm_instance()
+    if type_ in ['branch'] and vcs_repo.alias in ('hg', 'git'):
+        return vcs_repo.branches[name]
+    elif type_ in ['bookmark', 'book'] and vcs_repo.alias == 'hg':
+        return vcs_repo.bookmarks[name]
+    else:
+        raise ValueError()
+
+
+def resolve_ref_or_error(ref, repo, allowed_ref_types=None):
+    allowed_ref_types = allowed_ref_types or ['bookmark', 'book', 'tag', 'branch']
+
     def _parse_ref(type_, name, hash_=None):
         return type_, name, hash_
 
     try:
         ref_type, ref_name, ref_hash = _parse_ref(*ref.split(':'))
     except TypeError:
         raise JSONRPCError(
             'Ref `{ref}` given in a wrong format. Please check the API'
             ' documentation for more details'.format(ref=ref))
 
+    if ref_type not in allowed_ref_types:
+        raise JSONRPCError(
+            'Ref `{ref}` type is not allowed. '
+            'Only:{allowed_refs} are possible.'.format(
+                ref=ref, allowed_refs=allowed_ref_types))
+
     try:
         ref_hash = ref_hash or _get_ref_hash(repo, ref_type, ref_name)
     except (KeyError, ValueError):
         raise JSONRPCError(
             'The specified value:{type}:`{name}` does not exist, or is not allowed.'.format(
                 type=ref_type, name=ref_name))
 
     return ':'.join([ref_type, ref_name, ref_hash])
 
 
 def _get_commit_dict(
         filename, op, new_revision=None, old_revision=None,
         raw_diff=None, stats=None):
     if stats is None:
         stats = {
             "added": None,
             "binary": None,
             "deleted": None
         }
     return {
         "filename": safe_unicode(filename),
         "op": op,
 
         # extra details
         "new_revision": new_revision,
         "old_revision": old_revision,
 
         "raw_diff": raw_diff,
         "stats": stats
     }
-
-
-def _get_ref_hash(repo, type_, name):
-    vcs_repo = repo.scm_instance()
-    if type_ == 'branch' and vcs_repo.alias in ('hg', 'git'):
-        return vcs_repo.branches[name]
-    elif type_ == 'bookmark' and vcs_repo.alias == 'hg':
-        return vcs_repo.bookmarks[name]
-    else:
-        raise ValueError()

rhodecode/model/pull_request.py

@@ -1,1737 +1,1739 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2012-2018 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/
 
 
 """
 pull request model for RhodeCode
 """
 
 
 import json
 import logging
 import datetime
 import urllib
 import collections
 
 from pyramid.threadlocal import get_current_request
 
 from rhodecode import events
 from rhodecode.translation import lazy_ugettext#, _
 from rhodecode.lib import helpers as h, hooks_utils, diffs
 from rhodecode.lib import audit_logger
 from rhodecode.lib.compat import OrderedDict
 from rhodecode.lib.hooks_daemon import prepare_callback_daemon
 from rhodecode.lib.markup_renderer import (
     DEFAULT_COMMENTS_RENDERER, RstTemplateRenderer)
 from rhodecode.lib.utils2 import safe_unicode, safe_str, md5_safe
 from rhodecode.lib.vcs.backends.base import (
     Reference, MergeResponse, MergeFailureReason, UpdateFailureReason)
 from rhodecode.lib.vcs.conf import settings as vcs_settings
 from rhodecode.lib.vcs.exceptions import (
     CommitDoesNotExistError, EmptyRepositoryError)
 from rhodecode.model import BaseModel
 from rhodecode.model.changeset_status import ChangesetStatusModel
 from rhodecode.model.comment import CommentsModel
 from rhodecode.model.db import (
     or_, PullRequest, PullRequestReviewers, ChangesetStatus,
     PullRequestVersion, ChangesetComment, Repository, RepoReviewRule)
 from rhodecode.model.meta import Session
 from rhodecode.model.notification import NotificationModel, \
     EmailNotificationModel
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.settings import VcsSettingsModel
 
 
 log = logging.getLogger(__name__)
 
 
 # Data structure to hold the response data when updating commits during a pull
 # request update.
 UpdateResponse = collections.namedtuple('UpdateResponse', [
     'executed', 'reason', 'new', 'old', 'changes',
     'source_changed', 'target_changed'])
 
 
 class PullRequestModel(BaseModel):
 
     cls = PullRequest
 
     DIFF_CONTEXT = diffs.DEFAULT_CONTEXT
 
     MERGE_STATUS_MESSAGES = {
         MergeFailureReason.NONE: lazy_ugettext(
             'This pull request can be automatically merged.'),
         MergeFailureReason.UNKNOWN: lazy_ugettext(
             'This pull request cannot be merged because of an unhandled'
             ' exception.'),
         MergeFailureReason.MERGE_FAILED: lazy_ugettext(
             'This pull request cannot be merged because of merge conflicts.'),
         MergeFailureReason.PUSH_FAILED: lazy_ugettext(
             'This pull request could not be merged because push to target'
             ' failed.'),
         MergeFailureReason.TARGET_IS_NOT_HEAD: lazy_ugettext(
             'This pull request cannot be merged because the target is not a'
             ' head.'),
         MergeFailureReason.HG_SOURCE_HAS_MORE_BRANCHES: lazy_ugettext(
             'This pull request cannot be merged because the source contains'
             ' more branches than the target.'),
         MergeFailureReason.HG_TARGET_HAS_MULTIPLE_HEADS: lazy_ugettext(
             'This pull request cannot be merged because the target has'
             ' multiple heads.'),
         MergeFailureReason.TARGET_IS_LOCKED: lazy_ugettext(
             'This pull request cannot be merged because the target repository'
             ' is locked.'),
         MergeFailureReason._DEPRECATED_MISSING_COMMIT: lazy_ugettext(
             'This pull request cannot be merged because the target or the '
             'source reference is missing.'),
         MergeFailureReason.MISSING_TARGET_REF: lazy_ugettext(
             'This pull request cannot be merged because the target '
             'reference is missing.'),
         MergeFailureReason.MISSING_SOURCE_REF: lazy_ugettext(
             'This pull request cannot be merged because the source '
             'reference is missing.'),
         MergeFailureReason.SUBREPO_MERGE_FAILED: lazy_ugettext(
             'This pull request cannot be merged because of conflicts related '
             'to sub repositories.'),
     }
 
     UPDATE_STATUS_MESSAGES = {
         UpdateFailureReason.NONE: lazy_ugettext(
             'Pull request update successful.'),
         UpdateFailureReason.UNKNOWN: lazy_ugettext(
             'Pull request update failed because of an unknown error.'),
         UpdateFailureReason.NO_CHANGE: lazy_ugettext(
             'No update needed because the source and target have not changed.'),
         UpdateFailureReason.WRONG_REF_TYPE: lazy_ugettext(
             'Pull request cannot be updated because the reference type is '
             'not supported for an update. Only Branch, Tag or Bookmark is allowed.'),
         UpdateFailureReason.MISSING_TARGET_REF: lazy_ugettext(
             'This pull request cannot be updated because the target '
             'reference is missing.'),
         UpdateFailureReason.MISSING_SOURCE_REF: lazy_ugettext(
             'This pull request cannot be updated because the source '
             'reference is missing.'),
     }
+    REF_TYPES = ['bookmark', 'book', 'tag', 'branch']
+    UPDATABLE_REF_TYPES = ['bookmark', 'book', 'branch']
 
     def __get_pull_request(self, pull_request):
         return self._get_instance((
             PullRequest, PullRequestVersion), pull_request)
 
     def _check_perms(self, perms, pull_request, user, api=False):
         if not api:
             return h.HasRepoPermissionAny(*perms)(
                 user=user, repo_name=pull_request.target_repo.repo_name)
         else:
             return h.HasRepoPermissionAnyApi(*perms)(
                 user=user, repo_name=pull_request.target_repo.repo_name)
 
     def check_user_read(self, pull_request, user, api=False):
         _perms = ('repository.admin', 'repository.write', 'repository.read',)
         return self._check_perms(_perms, pull_request, user, api)
 
     def check_user_merge(self, pull_request, user, api=False):
         _perms = ('repository.admin', 'repository.write', 'hg.admin',)
         return self._check_perms(_perms, pull_request, user, api)
 
     def check_user_update(self, pull_request, user, api=False):
         owner = user.user_id == pull_request.user_id
         return self.check_user_merge(pull_request, user, api) or owner
 
     def check_user_delete(self, pull_request, user):
         owner = user.user_id == pull_request.user_id
         _perms = ('repository.admin',)
         return self._check_perms(_perms, pull_request, user) or owner
 
     def check_user_change_status(self, pull_request, user, api=False):
         reviewer = user.user_id in [x.user_id for x in
                                     pull_request.reviewers]
         return self.check_user_update(pull_request, user, api) or reviewer
 
     def check_user_comment(self, pull_request, user):
         owner = user.user_id == pull_request.user_id
         return self.check_user_read(pull_request, user) or owner
 
     def get(self, pull_request):
         return self.__get_pull_request(pull_request)
 
     def _prepare_get_all_query(self, repo_name, source=False, statuses=None,
                                opened_by=None, order_by=None,
                                order_dir='desc'):
         repo = None
         if repo_name:
             repo = self._get_repo(repo_name)
 
         q = PullRequest.query()
 
         # source or target
         if repo and source:
             q = q.filter(PullRequest.source_repo == repo)
         elif repo:
             q = q.filter(PullRequest.target_repo == repo)
 
         # closed,opened
         if statuses:
             q = q.filter(PullRequest.status.in_(statuses))
 
         # opened by filter
         if opened_by:
             q = q.filter(PullRequest.user_id.in_(opened_by))
 
         if order_by:
             order_map = {
                 'name_raw': PullRequest.pull_request_id,
                 'title': PullRequest.title,
                 'updated_on_raw': PullRequest.updated_on,
                 'target_repo': PullRequest.target_repo_id
             }
             if order_dir == 'asc':
                 q = q.order_by(order_map[order_by].asc())
             else:
                 q = q.order_by(order_map[order_by].desc())
 
         return q
 
     def count_all(self, repo_name, source=False, statuses=None,
                   opened_by=None):
         """
         Count the number of pull requests for a specific repository.
 
         :param repo_name: target or source repo
         :param source: boolean flag to specify if repo_name refers to source
         :param statuses: list of pull request statuses
         :param opened_by: author user of the pull request
         :returns: int number of pull requests
         """
         q = self._prepare_get_all_query(
             repo_name, source=source, statuses=statuses, opened_by=opened_by)
 
         return q.count()
 
     def get_all(self, repo_name, source=False, statuses=None, opened_by=None,
                 offset=0, length=None, order_by=None, order_dir='desc'):
         """
         Get all pull requests for a specific repository.
 
         :param repo_name: target or source repo
         :param source: boolean flag to specify if repo_name refers to source
         :param statuses: list of pull request statuses
         :param opened_by: author user of the pull request
         :param offset: pagination offset
         :param length: length of returned list
         :param order_by: order of the returned list
         :param order_dir: 'asc' or 'desc' ordering direction
         :returns: list of pull requests
         """
         q = self._prepare_get_all_query(
             repo_name, source=source, statuses=statuses, opened_by=opened_by,
             order_by=order_by, order_dir=order_dir)
 
         if length:
             pull_requests = q.limit(length).offset(offset).all()
         else:
             pull_requests = q.all()
 
         return pull_requests
 
     def count_awaiting_review(self, repo_name, source=False, statuses=None,
                               opened_by=None):
         """
         Count the number of pull requests for a specific repository that are
         awaiting review.
 
         :param repo_name: target or source repo
         :param source: boolean flag to specify if repo_name refers to source
         :param statuses: list of pull request statuses
         :param opened_by: author user of the pull request
         :returns: int number of pull requests
         """
         pull_requests = self.get_awaiting_review(
             repo_name, source=source, statuses=statuses, opened_by=opened_by)
 
         return len(pull_requests)
 
     def get_awaiting_review(self, repo_name, source=False, statuses=None,
                             opened_by=None, offset=0, length=None,
                             order_by=None, order_dir='desc'):
         """
         Get all pull requests for a specific repository that are awaiting
         review.
 
         :param repo_name: target or source repo
         :param source: boolean flag to specify if repo_name refers to source
         :param statuses: list of pull request statuses
         :param opened_by: author user of the pull request
         :param offset: pagination offset
         :param length: length of returned list
         :param order_by: order of the returned list
         :param order_dir: 'asc' or 'desc' ordering direction
         :returns: list of pull requests
         """
         pull_requests = self.get_all(
             repo_name, source=source, statuses=statuses, opened_by=opened_by,
             order_by=order_by, order_dir=order_dir)
 
         _filtered_pull_requests = []
         for pr in pull_requests:
             status = pr.calculated_review_status()
             if status in [ChangesetStatus.STATUS_NOT_REVIEWED,
                           ChangesetStatus.STATUS_UNDER_REVIEW]:
                 _filtered_pull_requests.append(pr)
         if length:
             return _filtered_pull_requests[offset:offset+length]
         else:
             return _filtered_pull_requests
 
     def count_awaiting_my_review(self, repo_name, source=False, statuses=None,
                                  opened_by=None, user_id=None):
         """
         Count the number of pull requests for a specific repository that are
         awaiting review from a specific user.
 
         :param repo_name: target or source repo
         :param source: boolean flag to specify if repo_name refers to source
         :param statuses: list of pull request statuses
         :param opened_by: author user of the pull request
         :param user_id: reviewer user of the pull request
         :returns: int number of pull requests
         """
         pull_requests = self.get_awaiting_my_review(
             repo_name, source=source, statuses=statuses, opened_by=opened_by,
             user_id=user_id)
 
         return len(pull_requests)
 
     def get_awaiting_my_review(self, repo_name, source=False, statuses=None,
                                opened_by=None, user_id=None, offset=0,
                                length=None, order_by=None, order_dir='desc'):
         """
         Get all pull requests for a specific repository that are awaiting
         review from a specific user.
 
         :param repo_name: target or source repo
         :param source: boolean flag to specify if repo_name refers to source
         :param statuses: list of pull request statuses
         :param opened_by: author user of the pull request
         :param user_id: reviewer user of the pull request
         :param offset: pagination offset
         :param length: length of returned list
         :param order_by: order of the returned list
         :param order_dir: 'asc' or 'desc' ordering direction
         :returns: list of pull requests
         """
         pull_requests = self.get_all(
             repo_name, source=source, statuses=statuses, opened_by=opened_by,
             order_by=order_by, order_dir=order_dir)
 
         _my = PullRequestModel().get_not_reviewed(user_id)
         my_participation = []
         for pr in pull_requests:
             if pr in _my:
                 my_participation.append(pr)
         _filtered_pull_requests = my_participation
         if length:
             return _filtered_pull_requests[offset:offset+length]
         else:
             return _filtered_pull_requests
 
     def get_not_reviewed(self, user_id):
         return [
             x.pull_request for x in PullRequestReviewers.query().filter(
                 PullRequestReviewers.user_id == user_id).all()
         ]
 
     def _prepare_participating_query(self, user_id=None, statuses=None,
                                      order_by=None, order_dir='desc'):
         q = PullRequest.query()
         if user_id:
             reviewers_subquery = Session().query(
                 PullRequestReviewers.pull_request_id).filter(
                 PullRequestReviewers.user_id == user_id).subquery()
             user_filter = or_(
                 PullRequest.user_id == user_id,
                 PullRequest.pull_request_id.in_(reviewers_subquery)
             )
             q = PullRequest.query().filter(user_filter)
 
         # closed,opened
         if statuses:
             q = q.filter(PullRequest.status.in_(statuses))
 
         if order_by:
             order_map = {
                 'name_raw': PullRequest.pull_request_id,
                 'title': PullRequest.title,
                 'updated_on_raw': PullRequest.updated_on,
                 'target_repo': PullRequest.target_repo_id
             }
             if order_dir == 'asc':
                 q = q.order_by(order_map[order_by].asc())
             else:
                 q = q.order_by(order_map[order_by].desc())
 
         return q
 
     def count_im_participating_in(self, user_id=None, statuses=None):
         q = self._prepare_participating_query(user_id, statuses=statuses)
         return q.count()
 
     def get_im_participating_in(
             self, user_id=None, statuses=None, offset=0,
             length=None, order_by=None, order_dir='desc'):
         """
         Get all Pull requests that i'm participating in, or i have opened
         """
 
         q = self._prepare_participating_query(
             user_id, statuses=statuses, order_by=order_by,
             order_dir=order_dir)
 
         if length:
             pull_requests = q.limit(length).offset(offset).all()
         else:
             pull_requests = q.all()
 
         return pull_requests
 
     def get_versions(self, pull_request):
         """
         returns version of pull request sorted by ID descending
         """
         return PullRequestVersion.query()\
             .filter(PullRequestVersion.pull_request == pull_request)\
             .order_by(PullRequestVersion.pull_request_version_id.asc())\
             .all()
 
     def get_pr_version(self, pull_request_id, version=None):
         at_version = None
 
         if version and version == 'latest':
             pull_request_ver = PullRequest.get(pull_request_id)
             pull_request_obj = pull_request_ver
             _org_pull_request_obj = pull_request_obj
             at_version = 'latest'
         elif version:
             pull_request_ver = PullRequestVersion.get_or_404(version)
             pull_request_obj = pull_request_ver
             _org_pull_request_obj = pull_request_ver.pull_request
             at_version = pull_request_ver.pull_request_version_id
         else:
             _org_pull_request_obj = pull_request_obj = PullRequest.get_or_404(
                 pull_request_id)
 
         pull_request_display_obj = PullRequest.get_pr_display_object(
             pull_request_obj, _org_pull_request_obj)
 
         return _org_pull_request_obj, pull_request_obj, \
                pull_request_display_obj, at_version
 
     def create(self, created_by, source_repo, source_ref, target_repo,
                target_ref, revisions, reviewers, title, description=None,
                description_renderer=None,
                reviewer_data=None, translator=None, auth_user=None):
         translator = translator or get_current_request().translate
 
         created_by_user = self._get_user(created_by)
         auth_user = auth_user or created_by_user.AuthUser()
         source_repo = self._get_repo(source_repo)
         target_repo = self._get_repo(target_repo)
 
         pull_request = PullRequest()
         pull_request.source_repo = source_repo
         pull_request.source_ref = source_ref
         pull_request.target_repo = target_repo
         pull_request.target_ref = target_ref
         pull_request.revisions = revisions
         pull_request.title = title
         pull_request.description = description
         pull_request.description_renderer = description_renderer
         pull_request.author = created_by_user
         pull_request.reviewer_data = reviewer_data
 
         Session().add(pull_request)
         Session().flush()
 
         reviewer_ids = set()
         # members / reviewers
         for reviewer_object in reviewers:
             user_id, reasons, mandatory, rules = reviewer_object
             user = self._get_user(user_id)
 
             # skip duplicates
             if user.user_id in reviewer_ids:
                 continue
 
             reviewer_ids.add(user.user_id)
 
             reviewer = PullRequestReviewers()
             reviewer.user = user
             reviewer.pull_request = pull_request
             reviewer.reasons = reasons
             reviewer.mandatory = mandatory
 
             # NOTE(marcink): pick only first rule for now
             rule_id = list(rules)[0] if rules else None
             rule = RepoReviewRule.get(rule_id) if rule_id else None
             if rule:
                 review_group = rule.user_group_vote_rule(user_id)
                 # we check if this particular reviewer is member of a voting group
                 if review_group:
                     # NOTE(marcink):
                     # can be that user is member of more but we pick the first same,
                     # same as default reviewers algo
                     review_group = review_group[0]
 
                     rule_data = {
                         'rule_name':
                             rule.review_rule_name,
                         'rule_user_group_entry_id':
                             review_group.repo_review_rule_users_group_id,
                         'rule_user_group_name':
                             review_group.users_group.users_group_name,
                         'rule_user_group_members':
                             [x.user.username for x in review_group.users_group.members],
                         'rule_user_group_members_id':
                             [x.user.user_id for x in review_group.users_group.members],
                     }
                     # e.g {'vote_rule': -1, 'mandatory': True}
                     rule_data.update(review_group.rule_data())
 
                     reviewer.rule_data = rule_data
 
             Session().add(reviewer)
             Session().flush()
 
         # Set approval status to "Under Review" for all commits which are
         # part of this pull request.
         ChangesetStatusModel().set_status(
             repo=target_repo,
             status=ChangesetStatus.STATUS_UNDER_REVIEW,
             user=created_by_user,
             pull_request=pull_request
         )
         # we commit early at this point. This has to do with a fact
         # that before queries do some row-locking. And because of that
         # we need to commit and finish transation before below validate call
         # that for large repos could be long resulting in long row locks
         Session().commit()
 
         # prepare workspace, and run initial merge simulation
         MergeCheck.validate(
             pull_request, auth_user=auth_user, translator=translator)
 
         self.notify_reviewers(pull_request, reviewer_ids)
         self._trigger_pull_request_hook(
             pull_request, created_by_user, 'create')
 
         creation_data = pull_request.get_api_data(with_merge_state=False)
         self._log_audit_action(
             'repo.pull_request.create', {'data': creation_data},
             auth_user, pull_request)
 
         return pull_request
 
     def _trigger_pull_request_hook(self, pull_request, user, action):
         pull_request = self.__get_pull_request(pull_request)
         target_scm = pull_request.target_repo.scm_instance()
         if action == 'create':
             trigger_hook = hooks_utils.trigger_log_create_pull_request_hook
         elif action == 'merge':
             trigger_hook = hooks_utils.trigger_log_merge_pull_request_hook
         elif action == 'close':
             trigger_hook = hooks_utils.trigger_log_close_pull_request_hook
         elif action == 'review_status_change':
             trigger_hook = hooks_utils.trigger_log_review_pull_request_hook
         elif action == 'update':
             trigger_hook = hooks_utils.trigger_log_update_pull_request_hook
         else:
             return
 
         trigger_hook(
             username=user.username,
             repo_name=pull_request.target_repo.repo_name,
             repo_alias=target_scm.alias,
             pull_request=pull_request)
 
     def _get_commit_ids(self, pull_request):
         """
         Return the commit ids of the merged pull request.
 
         This method is not dealing correctly yet with the lack of autoupdates
         nor with the implicit target updates.
         For example: if a commit in the source repo is already in the target it
         will be reported anyways.
         """
         merge_rev = pull_request.merge_rev
         if merge_rev is None:
             raise ValueError('This pull request was not merged yet')
 
         commit_ids = list(pull_request.revisions)
         if merge_rev not in commit_ids:
             commit_ids.append(merge_rev)
 
         return commit_ids
 
     def merge_repo(self, pull_request, user, extras):
         log.debug("Merging pull request %s", pull_request.pull_request_id)
         extras['user_agent'] = 'internal-merge'
         merge_state = self._merge_pull_request(pull_request, user, extras)
         if merge_state.executed:
             log.debug(
                 "Merge was successful, updating the pull request comments.")
             self._comment_and_close_pr(pull_request, user, merge_state)
 
             self._log_audit_action(
                 'repo.pull_request.merge',
                 {'merge_state': merge_state.__dict__},
                 user, pull_request)
 
         else:
             log.warn("Merge failed, not updating the pull request.")
         return merge_state
 
     def _merge_pull_request(self, pull_request, user, extras, merge_msg=None):
         target_vcs = pull_request.target_repo.scm_instance()
         source_vcs = pull_request.source_repo.scm_instance()
 
         message = safe_unicode(merge_msg or vcs_settings.MERGE_MESSAGE_TMPL).format(
             pr_id=pull_request.pull_request_id,
             pr_title=pull_request.title,
             source_repo=source_vcs.name,
             source_ref_name=pull_request.source_ref_parts.name,
             target_repo=target_vcs.name,
             target_ref_name=pull_request.target_ref_parts.name,
         )
 
         workspace_id = self._workspace_id(pull_request)
         repo_id = pull_request.target_repo.repo_id
         use_rebase = self._use_rebase_for_merging(pull_request)
         close_branch = self._close_branch_before_merging(pull_request)
 
         target_ref = self._refresh_reference(
             pull_request.target_ref_parts, target_vcs)
 
         callback_daemon, extras = prepare_callback_daemon(
             extras, protocol=vcs_settings.HOOKS_PROTOCOL,
             host=vcs_settings.HOOKS_HOST,
             use_direct_calls=vcs_settings.HOOKS_DIRECT_CALLS)
 
         with callback_daemon:
             # TODO: johbo: Implement a clean way to run a config_override
             # for a single call.
             target_vcs.config.set(
                 'rhodecode', 'RC_SCM_DATA', json.dumps(extras))
 
             user_name = user.short_contact
             merge_state = target_vcs.merge(
                 repo_id, workspace_id, target_ref, source_vcs,
                 pull_request.source_ref_parts,
                 user_name=user_name, user_email=user.email,
                 message=message, use_rebase=use_rebase,
                 close_branch=close_branch)
         return merge_state
 
     def _comment_and_close_pr(self, pull_request, user, merge_state, close_msg=None):
         pull_request.merge_rev = merge_state.merge_ref.commit_id
         pull_request.updated_on = datetime.datetime.now()
         close_msg = close_msg or 'Pull request merged and closed'
 
         CommentsModel().create(
             text=safe_unicode(close_msg),
             repo=pull_request.target_repo.repo_id,
             user=user.user_id,
             pull_request=pull_request.pull_request_id,
             f_path=None,
             line_no=None,
             closing_pr=True
         )
 
         Session().add(pull_request)
         Session().flush()
         # TODO: paris: replace invalidation with less radical solution
         ScmModel().mark_for_invalidation(
             pull_request.target_repo.repo_name)
         self._trigger_pull_request_hook(pull_request, user, 'merge')
 
     def has_valid_update_type(self, pull_request):
         source_ref_type = pull_request.source_ref_parts.type
-        return source_ref_type in ['book', 'branch', 'tag']
+        return source_ref_type in self.REF_TYPES
 
     def update_commits(self, pull_request):
         """
         Get the updated list of commits for the pull request
         and return the new pull request version and the list
         of commits processed by this update action
         """
         pull_request = self.__get_pull_request(pull_request)
         source_ref_type = pull_request.source_ref_parts.type
         source_ref_name = pull_request.source_ref_parts.name
         source_ref_id = pull_request.source_ref_parts.commit_id
 
         target_ref_type = pull_request.target_ref_parts.type
         target_ref_name = pull_request.target_ref_parts.name
         target_ref_id = pull_request.target_ref_parts.commit_id
 
         if not self.has_valid_update_type(pull_request):
             log.debug(
                 "Skipping update of pull request %s due to ref type: %s",
                 pull_request, source_ref_type)
             return UpdateResponse(
                 executed=False,
                 reason=UpdateFailureReason.WRONG_REF_TYPE,
                 old=pull_request, new=None, changes=None,
                 source_changed=False, target_changed=False)
 
         # source repo
         source_repo = pull_request.source_repo.scm_instance()
         try:
             source_commit = source_repo.get_commit(commit_id=source_ref_name)
         except CommitDoesNotExistError:
             return UpdateResponse(
                 executed=False,
                 reason=UpdateFailureReason.MISSING_SOURCE_REF,
                 old=pull_request, new=None, changes=None,
                 source_changed=False, target_changed=False)
 
         source_changed = source_ref_id != source_commit.raw_id
 
         # target repo
         target_repo = pull_request.target_repo.scm_instance()
         try:
             target_commit = target_repo.get_commit(commit_id=target_ref_name)
         except CommitDoesNotExistError:
             return UpdateResponse(
                 executed=False,
                 reason=UpdateFailureReason.MISSING_TARGET_REF,
                 old=pull_request, new=None, changes=None,
                 source_changed=False, target_changed=False)
         target_changed = target_ref_id != target_commit.raw_id
 
         if not (source_changed or target_changed):
             log.debug("Nothing changed in pull request %s", pull_request)
             return UpdateResponse(
                 executed=False,
                 reason=UpdateFailureReason.NO_CHANGE,
                 old=pull_request, new=None, changes=None,
                 source_changed=target_changed, target_changed=source_changed)
 
         change_in_found = 'target repo' if target_changed else 'source repo'
         log.debug('Updating pull request because of change in %s detected',
                   change_in_found)
 
         # Finally there is a need for an update, in case of source change
         # we create a new version, else just an update
         if source_changed:
             pull_request_version = self._create_version_from_snapshot(pull_request)
             self._link_comments_to_version(pull_request_version)
         else:
             try:
                 ver = pull_request.versions[-1]
             except IndexError:
                 ver = None
 
             pull_request.pull_request_version_id = \
                 ver.pull_request_version_id if ver else None
             pull_request_version = pull_request
 
         try:
-            if target_ref_type in ('tag', 'branch', 'book'):
+            if target_ref_type in self.REF_TYPES:
                 target_commit = target_repo.get_commit(target_ref_name)
             else:
                 target_commit = target_repo.get_commit(target_ref_id)
         except CommitDoesNotExistError:
             return UpdateResponse(
                 executed=False,
                 reason=UpdateFailureReason.MISSING_TARGET_REF,
                 old=pull_request, new=None, changes=None,
                 source_changed=source_changed, target_changed=target_changed)
 
         # re-compute commit ids
         old_commit_ids = pull_request.revisions
         pre_load = ["author", "branch", "date", "message"]
         commit_ranges = target_repo.compare(
             target_commit.raw_id, source_commit.raw_id, source_repo, merge=True,
             pre_load=pre_load)
 
         ancestor = target_repo.get_common_ancestor(
             target_commit.raw_id, source_commit.raw_id, source_repo)
 
         pull_request.source_ref = '%s:%s:%s' % (
             source_ref_type, source_ref_name, source_commit.raw_id)
         pull_request.target_ref = '%s:%s:%s' % (
             target_ref_type, target_ref_name, ancestor)
 
         pull_request.revisions = [
             commit.raw_id for commit in reversed(commit_ranges)]
         pull_request.updated_on = datetime.datetime.now()
         Session().add(pull_request)
         new_commit_ids = pull_request.revisions
 
         old_diff_data, new_diff_data = self._generate_update_diffs(
             pull_request, pull_request_version)
 
         # calculate commit and file changes
         changes = self._calculate_commit_id_changes(
             old_commit_ids, new_commit_ids)
         file_changes = self._calculate_file_changes(
             old_diff_data, new_diff_data)
 
         # set comments as outdated if DIFFS changed
         CommentsModel().outdate_comments(
             pull_request, old_diff_data=old_diff_data,
             new_diff_data=new_diff_data)
 
         commit_changes = (changes.added or changes.removed)
         file_node_changes = (
             file_changes.added or file_changes.modified or file_changes.removed)
         pr_has_changes = commit_changes or file_node_changes
 
         # Add an automatic comment to the pull request, in case
         # anything has changed
         if pr_has_changes:
             update_comment = CommentsModel().create(
                 text=self._render_update_message(changes, file_changes),
                 repo=pull_request.target_repo,
                 user=pull_request.author,
                 pull_request=pull_request,
                 send_email=False, renderer=DEFAULT_COMMENTS_RENDERER)
 
             # Update status to "Under Review" for added commits
             for commit_id in changes.added:
                 ChangesetStatusModel().set_status(
                     repo=pull_request.source_repo,
                     status=ChangesetStatus.STATUS_UNDER_REVIEW,
                     comment=update_comment,
                     user=pull_request.author,
                     pull_request=pull_request,
                     revision=commit_id)
 
         log.debug(
             'Updated pull request %s, added_ids: %s, common_ids: %s, '
             'removed_ids: %s', pull_request.pull_request_id,
             changes.added, changes.common, changes.removed)
         log.debug(
             'Updated pull request with the following file changes: %s',
             file_changes)
 
         log.info(
             "Updated pull request %s from commit %s to commit %s, "
             "stored new version %s of this pull request.",
             pull_request.pull_request_id, source_ref_id,
             pull_request.source_ref_parts.commit_id,
             pull_request_version.pull_request_version_id)
         Session().commit()
         self._trigger_pull_request_hook(
             pull_request, pull_request.author, 'update')
 
         return UpdateResponse(
             executed=True, reason=UpdateFailureReason.NONE,
             old=pull_request, new=pull_request_version, changes=changes,
             source_changed=source_changed, target_changed=target_changed)
 
     def _create_version_from_snapshot(self, pull_request):
         version = PullRequestVersion()
         version.title = pull_request.title
         version.description = pull_request.description
         version.status = pull_request.status
         version.created_on = datetime.datetime.now()
         version.updated_on = pull_request.updated_on
         version.user_id = pull_request.user_id
         version.source_repo = pull_request.source_repo
         version.source_ref = pull_request.source_ref
         version.target_repo = pull_request.target_repo
         version.target_ref = pull_request.target_ref
 
         version._last_merge_source_rev = pull_request._last_merge_source_rev
         version._last_merge_target_rev = pull_request._last_merge_target_rev
         version.last_merge_status = pull_request.last_merge_status
         version.shadow_merge_ref = pull_request.shadow_merge_ref
         version.merge_rev = pull_request.merge_rev
         version.reviewer_data = pull_request.reviewer_data
 
         version.revisions = pull_request.revisions
         version.pull_request = pull_request
         Session().add(version)
         Session().flush()
 
         return version
 
     def _generate_update_diffs(self, pull_request, pull_request_version):
 
         diff_context = (
             self.DIFF_CONTEXT +
             CommentsModel.needed_extra_diff_context())
         hide_whitespace_changes = False
         source_repo = pull_request_version.source_repo
         source_ref_id = pull_request_version.source_ref_parts.commit_id
         target_ref_id = pull_request_version.target_ref_parts.commit_id
         old_diff = self._get_diff_from_pr_or_version(
             source_repo, source_ref_id, target_ref_id,
             hide_whitespace_changes=hide_whitespace_changes, diff_context=diff_context)
 
         source_repo = pull_request.source_repo
         source_ref_id = pull_request.source_ref_parts.commit_id
         target_ref_id = pull_request.target_ref_parts.commit_id
 
         new_diff = self._get_diff_from_pr_or_version(
             source_repo, source_ref_id, target_ref_id,
             hide_whitespace_changes=hide_whitespace_changes, diff_context=diff_context)
 
         old_diff_data = diffs.DiffProcessor(old_diff)
         old_diff_data.prepare()
         new_diff_data = diffs.DiffProcessor(new_diff)
         new_diff_data.prepare()
 
         return old_diff_data, new_diff_data
 
     def _link_comments_to_version(self, pull_request_version):
         """
         Link all unlinked comments of this pull request to the given version.
 
         :param pull_request_version: The `PullRequestVersion` to which
             the comments shall be linked.
 
         """
         pull_request = pull_request_version.pull_request
         comments = ChangesetComment.query()\
             .filter(
                 # TODO: johbo: Should we query for the repo at all here?
                 # Pending decision on how comments of PRs are to be related
                 # to either the source repo, the target repo or no repo at all.
                 ChangesetComment.repo_id == pull_request.target_repo.repo_id,
                 ChangesetComment.pull_request == pull_request,
                 ChangesetComment.pull_request_version == None)\
             .order_by(ChangesetComment.comment_id.asc())
 
         # TODO: johbo: Find out why this breaks if it is done in a bulk
         # operation.
         for comment in comments:
             comment.pull_request_version_id = (
                 pull_request_version.pull_request_version_id)
             Session().add(comment)
 
     def _calculate_commit_id_changes(self, old_ids, new_ids):
         added = [x for x in new_ids if x not in old_ids]
         common = [x for x in new_ids if x in old_ids]
         removed = [x for x in old_ids if x not in new_ids]
         total = new_ids
         return ChangeTuple(added, common, removed, total)
 
     def _calculate_file_changes(self, old_diff_data, new_diff_data):
 
         old_files = OrderedDict()
         for diff_data in old_diff_data.parsed_diff:
             old_files[diff_data['filename']] = md5_safe(diff_data['raw_diff'])
 
         added_files = []
         modified_files = []
         removed_files = []
         for diff_data in new_diff_data.parsed_diff:
             new_filename = diff_data['filename']
             new_hash = md5_safe(diff_data['raw_diff'])
 
             old_hash = old_files.get(new_filename)
             if not old_hash:
                 # file is not present in old diff, means it's added
                 added_files.append(new_filename)
             else:
                 if new_hash != old_hash:
                     modified_files.append(new_filename)
                 # now remove a file from old, since we have seen it already
                 del old_files[new_filename]
 
         # removed files is when there are present in old, but not in NEW,
         # since we remove old files that are present in new diff, left-overs
         # if any should be the removed files
         removed_files.extend(old_files.keys())
 
         return FileChangeTuple(added_files, modified_files, removed_files)
 
     def _render_update_message(self, changes, file_changes):
         """
         render the message using DEFAULT_COMMENTS_RENDERER (RST renderer),
         so it's always looking the same disregarding on which default
         renderer system is using.
 
         :param changes: changes named tuple
         :param file_changes: file changes named tuple
 
         """
         new_status = ChangesetStatus.get_status_lbl(
             ChangesetStatus.STATUS_UNDER_REVIEW)
 
         changed_files = (
             file_changes.added + file_changes.modified + file_changes.removed)
 
         params = {
             'under_review_label': new_status,
             'added_commits': changes.added,
             'removed_commits': changes.removed,
             'changed_files': changed_files,
             'added_files': file_changes.added,
             'modified_files': file_changes.modified,
             'removed_files': file_changes.removed,
         }
         renderer = RstTemplateRenderer()
         return renderer.render('pull_request_update.mako', **params)
 
     def edit(self, pull_request, title, description, description_renderer, user):
         pull_request = self.__get_pull_request(pull_request)
         old_data = pull_request.get_api_data(with_merge_state=False)
         if pull_request.is_closed():
             raise ValueError('This pull request is closed')
         if title:
             pull_request.title = title
         pull_request.description = description
         pull_request.updated_on = datetime.datetime.now()
         pull_request.description_renderer = description_renderer
         Session().add(pull_request)
         self._log_audit_action(
             'repo.pull_request.edit', {'old_data': old_data},
             user, pull_request)
 
     def update_reviewers(self, pull_request, reviewer_data, user):
         """
         Update the reviewers in the pull request
 
         :param pull_request: the pr to update
         :param reviewer_data: list of tuples
             [(user, ['reason1', 'reason2'], mandatory_flag, [rules])]
         """
         pull_request = self.__get_pull_request(pull_request)
         if pull_request.is_closed():
             raise ValueError('This pull request is closed')
 
         reviewers = {}
         for user_id, reasons, mandatory, rules in reviewer_data:
             if isinstance(user_id, (int, basestring)):
                 user_id = self._get_user(user_id).user_id
             reviewers[user_id] = {
                 'reasons': reasons, 'mandatory': mandatory}
 
         reviewers_ids = set(reviewers.keys())
         current_reviewers = PullRequestReviewers.query()\
             .filter(PullRequestReviewers.pull_request ==
                     pull_request).all()
         current_reviewers_ids = set([x.user.user_id for x in current_reviewers])
 
         ids_to_add = reviewers_ids.difference(current_reviewers_ids)
         ids_to_remove = current_reviewers_ids.difference(reviewers_ids)
 
         log.debug("Adding %s reviewers", ids_to_add)
         log.debug("Removing %s reviewers", ids_to_remove)
         changed = False
         for uid in ids_to_add:
             changed = True
             _usr = self._get_user(uid)
             reviewer = PullRequestReviewers()
             reviewer.user = _usr
             reviewer.pull_request = pull_request
             reviewer.reasons = reviewers[uid]['reasons']
             # NOTE(marcink): mandatory shouldn't be changed now
             # reviewer.mandatory = reviewers[uid]['reasons']
             Session().add(reviewer)
             self._log_audit_action(
                 'repo.pull_request.reviewer.add', {'data': reviewer.get_dict()},
                 user, pull_request)
 
         for uid in ids_to_remove:
             changed = True
             reviewers = PullRequestReviewers.query()\
                 .filter(PullRequestReviewers.user_id == uid,
                         PullRequestReviewers.pull_request == pull_request)\
                 .all()
             # use .all() in case we accidentally added the same person twice
             # this CAN happen due to the lack of DB checks
             for obj in reviewers:
                 old_data = obj.get_dict()
                 Session().delete(obj)
                 self._log_audit_action(
                     'repo.pull_request.reviewer.delete',
                     {'old_data': old_data}, user, pull_request)
 
         if changed:
             pull_request.updated_on = datetime.datetime.now()
             Session().add(pull_request)
 
         self.notify_reviewers(pull_request, ids_to_add)
         return ids_to_add, ids_to_remove
 
     def get_url(self, pull_request, request=None, permalink=False):
         if not request:
             request = get_current_request()
 
         if permalink:
             return request.route_url(
                 'pull_requests_global',
                 pull_request_id=pull_request.pull_request_id,)
         else:
             return request.route_url('pullrequest_show',
                 repo_name=safe_str(pull_request.target_repo.repo_name),
                 pull_request_id=pull_request.pull_request_id,)
 
     def get_shadow_clone_url(self, pull_request, request=None):
         """
         Returns qualified url pointing to the shadow repository. If this pull
         request is closed there is no shadow repository and ``None`` will be
         returned.
         """
         if pull_request.is_closed():
             return None
         else:
             pr_url = urllib.unquote(self.get_url(pull_request, request=request))
             return safe_unicode('{pr_url}/repository'.format(pr_url=pr_url))
 
     def notify_reviewers(self, pull_request, reviewers_ids):
         # notification to reviewers
         if not reviewers_ids:
             return
 
         pull_request_obj = pull_request
         # get the current participants of this pull request
         recipients = reviewers_ids
         notification_type = EmailNotificationModel.TYPE_PULL_REQUEST
 
         pr_source_repo = pull_request_obj.source_repo
         pr_target_repo = pull_request_obj.target_repo
 
         pr_url = h.route_url('pullrequest_show',
             repo_name=pr_target_repo.repo_name,
             pull_request_id=pull_request_obj.pull_request_id,)
 
         # set some variables for email notification
         pr_target_repo_url = h.route_url(
             'repo_summary', repo_name=pr_target_repo.repo_name)
 
         pr_source_repo_url = h.route_url(
             'repo_summary', repo_name=pr_source_repo.repo_name)
 
         # pull request specifics
         pull_request_commits = [
             (x.raw_id, x.message)
             for x in map(pr_source_repo.get_commit, pull_request.revisions)]
 
         kwargs = {
             'user': pull_request.author,
             'pull_request': pull_request_obj,
             'pull_request_commits': pull_request_commits,
 
             'pull_request_target_repo': pr_target_repo,
             'pull_request_target_repo_url': pr_target_repo_url,
 
             'pull_request_source_repo': pr_source_repo,
             'pull_request_source_repo_url': pr_source_repo_url,
 
             'pull_request_url': pr_url,
         }
 
         # pre-generate the subject for notification itself
         (subject,
          _h, _e,  # we don't care about those
          body_plaintext) = EmailNotificationModel().render_email(
             notification_type, **kwargs)
 
         # create notification objects, and emails
         NotificationModel().create(
             created_by=pull_request.author,
             notification_subject=subject,
             notification_body=body_plaintext,
             notification_type=notification_type,
             recipients=recipients,
             email_kwargs=kwargs,
         )
 
     def delete(self, pull_request, user):
         pull_request = self.__get_pull_request(pull_request)
         old_data = pull_request.get_api_data(with_merge_state=False)
         self._cleanup_merge_workspace(pull_request)
         self._log_audit_action(
             'repo.pull_request.delete', {'old_data': old_data},
             user, pull_request)
         Session().delete(pull_request)
 
     def close_pull_request(self, pull_request, user):
         pull_request = self.__get_pull_request(pull_request)
         self._cleanup_merge_workspace(pull_request)
         pull_request.status = PullRequest.STATUS_CLOSED
         pull_request.updated_on = datetime.datetime.now()
         Session().add(pull_request)
         self._trigger_pull_request_hook(
             pull_request, pull_request.author, 'close')
 
         pr_data = pull_request.get_api_data(with_merge_state=False)
         self._log_audit_action(
             'repo.pull_request.close', {'data': pr_data}, user, pull_request)
 
     def close_pull_request_with_comment(
             self, pull_request, user, repo, message=None, auth_user=None):
 
         pull_request_review_status = pull_request.calculated_review_status()
 
         if pull_request_review_status == ChangesetStatus.STATUS_APPROVED:
             # approved only if we have voting consent
             status = ChangesetStatus.STATUS_APPROVED
         else:
             status = ChangesetStatus.STATUS_REJECTED
         status_lbl = ChangesetStatus.get_status_lbl(status)
 
         default_message = (
             'Closing with status change {transition_icon} {status}.'
         ).format(transition_icon='>', status=status_lbl)
         text = message or default_message
 
         # create a comment, and link it to new status
         comment = CommentsModel().create(
             text=text,
             repo=repo.repo_id,
             user=user.user_id,
             pull_request=pull_request.pull_request_id,
             status_change=status_lbl,
             status_change_type=status,
             closing_pr=True,
             auth_user=auth_user,
         )
 
         # calculate old status before we change it
         old_calculated_status = pull_request.calculated_review_status()
         ChangesetStatusModel().set_status(
             repo.repo_id,
             status,
             user.user_id,
             comment=comment,
             pull_request=pull_request.pull_request_id
         )
 
         Session().flush()
         events.trigger(events.PullRequestCommentEvent(pull_request, comment))
         # we now calculate the status of pull request again, and based on that
         # calculation trigger status change. This might happen in cases
         # that non-reviewer admin closes a pr, which means his vote doesn't
         # change the status, while if he's a reviewer this might change it.
         calculated_status = pull_request.calculated_review_status()
         if old_calculated_status != calculated_status:
             self._trigger_pull_request_hook(
                 pull_request, user, 'review_status_change')
 
         # finally close the PR
         PullRequestModel().close_pull_request(
             pull_request.pull_request_id, user)
 
         return comment, status
 
     def merge_status(self, pull_request, translator=None,
                      force_shadow_repo_refresh=False):
         _ = translator or get_current_request().translate
 
         if not self._is_merge_enabled(pull_request):
             return False, _('Server-side pull request merging is disabled.')
         if pull_request.is_closed():
             return False, _('This pull request is closed.')
         merge_possible, msg = self._check_repo_requirements(
             target=pull_request.target_repo, source=pull_request.source_repo,
             translator=_)
         if not merge_possible:
             return merge_possible, msg
 
         try:
             resp = self._try_merge(
                 pull_request,
                 force_shadow_repo_refresh=force_shadow_repo_refresh)
             log.debug("Merge response: %s", resp)
             status = resp.possible, self.merge_status_message(
                 resp.failure_reason)
         except NotImplementedError:
             status = False, _('Pull request merging is not supported.')
 
         return status
 
     def _check_repo_requirements(self, target, source, translator):
         """
         Check if `target` and `source` have compatible requirements.
 
         Currently this is just checking for largefiles.
         """
         _ = translator
         target_has_largefiles = self._has_largefiles(target)
         source_has_largefiles = self._has_largefiles(source)
         merge_possible = True
         message = u''
 
         if target_has_largefiles != source_has_largefiles:
             merge_possible = False
             if source_has_largefiles:
                 message = _(
                     'Target repository large files support is disabled.')
             else:
                 message = _(
                     'Source repository large files support is disabled.')
 
         return merge_possible, message
 
     def _has_largefiles(self, repo):
         largefiles_ui = VcsSettingsModel(repo=repo).get_ui_settings(
             'extensions', 'largefiles')
         return largefiles_ui and largefiles_ui[0].active
 
     def _try_merge(self, pull_request, force_shadow_repo_refresh=False):
         """
         Try to merge the pull request and return the merge status.
         """
         log.debug(
             "Trying out if the pull request %s can be merged. Force_refresh=%s",
             pull_request.pull_request_id, force_shadow_repo_refresh)
         target_vcs = pull_request.target_repo.scm_instance()
 
         # Refresh the target reference.
         try:
             target_ref = self._refresh_reference(
                 pull_request.target_ref_parts, target_vcs)
         except CommitDoesNotExistError:
             merge_state = MergeResponse(
                 False, False, None, MergeFailureReason.MISSING_TARGET_REF)
             return merge_state
 
         target_locked = pull_request.target_repo.locked
         if target_locked and target_locked[0]:
             log.debug("The target repository is locked.")
             merge_state = MergeResponse(
                 False, False, None, MergeFailureReason.TARGET_IS_LOCKED)
         elif force_shadow_repo_refresh or self._needs_merge_state_refresh(
                 pull_request, target_ref):
             log.debug("Refreshing the merge status of the repository.")
             merge_state = self._refresh_merge_state(
                 pull_request, target_vcs, target_ref)
         else:
             possible = pull_request.\
                 last_merge_status == MergeFailureReason.NONE
             merge_state = MergeResponse(
                 possible, False, None, pull_request.last_merge_status)
 
         return merge_state
 
     def _refresh_reference(self, reference, vcs_repository):
-        if reference.type in ('branch', 'book'):
+        if reference.type in self.UPDATABLE_REF_TYPES:
             name_or_id = reference.name
         else:
             name_or_id = reference.commit_id
         refreshed_commit = vcs_repository.get_commit(name_or_id)
         refreshed_reference = Reference(
             reference.type, reference.name, refreshed_commit.raw_id)
         return refreshed_reference
 
     def _needs_merge_state_refresh(self, pull_request, target_reference):
         return not(
             pull_request.revisions and
             pull_request.revisions[0] == pull_request._last_merge_source_rev and
             target_reference.commit_id == pull_request._last_merge_target_rev)
 
     def _refresh_merge_state(self, pull_request, target_vcs, target_reference):
         workspace_id = self._workspace_id(pull_request)
         source_vcs = pull_request.source_repo.scm_instance()
         repo_id = pull_request.target_repo.repo_id
         use_rebase = self._use_rebase_for_merging(pull_request)
         close_branch = self._close_branch_before_merging(pull_request)
         merge_state = target_vcs.merge(
             repo_id, workspace_id,
             target_reference, source_vcs, pull_request.source_ref_parts,
             dry_run=True, use_rebase=use_rebase,
             close_branch=close_branch)
 
         # Do not store the response if there was an unknown error.
         if merge_state.failure_reason != MergeFailureReason.UNKNOWN:
             pull_request._last_merge_source_rev = \
                 pull_request.source_ref_parts.commit_id
             pull_request._last_merge_target_rev = target_reference.commit_id
             pull_request.last_merge_status = merge_state.failure_reason
             pull_request.shadow_merge_ref = merge_state.merge_ref
             Session().add(pull_request)
             Session().commit()
 
         return merge_state
 
     def _workspace_id(self, pull_request):
         workspace_id = 'pr-%s' % pull_request.pull_request_id
         return workspace_id
 
     def merge_status_message(self, status_code):
         """
         Return a human friendly error message for the given merge status code.
         """
         return self.MERGE_STATUS_MESSAGES[status_code]
 
     def generate_repo_data(self, repo, commit_id=None, branch=None,
                            bookmark=None, translator=None):
         from rhodecode.model.repo import RepoModel
 
         all_refs, selected_ref = \
             self._get_repo_pullrequest_sources(
                 repo.scm_instance(), commit_id=commit_id,
                 branch=branch, bookmark=bookmark, translator=translator)
 
         refs_select2 = []
         for element in all_refs:
             children = [{'id': x[0], 'text': x[1]} for x in element[0]]
             refs_select2.append({'text': element[1], 'children': children})
 
         return {
             'user': {
                 'user_id': repo.user.user_id,
                 'username': repo.user.username,
                 'firstname': repo.user.first_name,
                 'lastname': repo.user.last_name,
                 'gravatar_link': h.gravatar_url(repo.user.email, 14),
             },
             'name': repo.repo_name,
             'link': RepoModel().get_url(repo),
             'description': h.chop_at_smart(repo.description_safe, '\n'),
             'refs': {
                 'all_refs': all_refs,
                 'selected_ref': selected_ref,
                 'select2_refs': refs_select2
             }
         }
 
     def generate_pullrequest_title(self, source, source_ref, target):
         return u'{source}#{at_ref} to {target}'.format(
             source=source,
             at_ref=source_ref,
             target=target,
         )
 
     def _cleanup_merge_workspace(self, pull_request):
         # Merging related cleanup
         repo_id = pull_request.target_repo.repo_id
         target_scm = pull_request.target_repo.scm_instance()
         workspace_id = self._workspace_id(pull_request)
 
         try:
             target_scm.cleanup_merge_workspace(repo_id, workspace_id)
         except NotImplementedError:
             pass
 
     def _get_repo_pullrequest_sources(
             self, repo, commit_id=None, branch=None, bookmark=None,
             translator=None):
         """
         Return a structure with repo's interesting commits, suitable for
         the selectors in pullrequest controller
 
         :param commit_id: a commit that must be in the list somehow
             and selected by default
         :param branch: a branch that must be in the list and selected
             by default - even if closed
         :param bookmark: a bookmark that must be in the list and selected
         """
         _ = translator or get_current_request().translate
 
         commit_id = safe_str(commit_id) if commit_id else None
         branch = safe_str(branch) if branch else None
         bookmark = safe_str(bookmark) if bookmark else None
 
         selected = None
 
         # order matters: first source that has commit_id in it will be selected
         sources = []
         sources.append(('book', repo.bookmarks.items(), _('Bookmarks'), bookmark))
         sources.append(('branch', repo.branches.items(), _('Branches'), branch))
 
         if commit_id:
             ref_commit = (h.short_id(commit_id), commit_id)
             sources.append(('rev', [ref_commit], _('Commit IDs'), commit_id))
 
         sources.append(
             ('branch', repo.branches_closed.items(), _('Closed Branches'), branch),
         )
 
         groups = []
         for group_key, ref_list, group_name, match in sources:
             group_refs = []
             for ref_name, ref_id in ref_list:
                 ref_key = '%s:%s:%s' % (group_key, ref_name, ref_id)
                 group_refs.append((ref_key, ref_name))
 
                 if not selected:
                     if set([commit_id, match]) & set([ref_id, ref_name]):
                         selected = ref_key
 
             if group_refs:
                 groups.append((group_refs, group_name))
 
         if not selected:
             ref = commit_id or branch or bookmark
             if ref:
                 raise CommitDoesNotExistError(
                     'No commit refs could be found matching: %s' % ref)
             elif repo.DEFAULT_BRANCH_NAME in repo.branches:
                 selected = 'branch:%s:%s' % (
                     repo.DEFAULT_BRANCH_NAME,
                     repo.branches[repo.DEFAULT_BRANCH_NAME]
                 )
             elif repo.commit_ids:
                 # make the user select in this case
                 selected = None
             else:
                 raise EmptyRepositoryError()
         return groups, selected
 
     def get_diff(self, source_repo, source_ref_id, target_ref_id,
                  hide_whitespace_changes, diff_context):
 
         return self._get_diff_from_pr_or_version(
             source_repo, source_ref_id, target_ref_id,
             hide_whitespace_changes=hide_whitespace_changes, diff_context=diff_context)
 
     def _get_diff_from_pr_or_version(
             self, source_repo, source_ref_id, target_ref_id,
             hide_whitespace_changes, diff_context):
 
         target_commit = source_repo.get_commit(
             commit_id=safe_str(target_ref_id))
         source_commit = source_repo.get_commit(
             commit_id=safe_str(source_ref_id))
         if isinstance(source_repo, Repository):
             vcs_repo = source_repo.scm_instance()
         else:
             vcs_repo = source_repo
 
         # TODO: johbo: In the context of an update, we cannot reach
         # the old commit anymore with our normal mechanisms. It needs
         # some sort of special support in the vcs layer to avoid this
         # workaround.
         if (source_commit.raw_id == vcs_repo.EMPTY_COMMIT_ID and
                 vcs_repo.alias == 'git'):
             source_commit.raw_id = safe_str(source_ref_id)
 
         log.debug('calculating diff between '
                   'source_ref:%s and target_ref:%s for repo `%s`',
                   target_ref_id, source_ref_id,
                   safe_unicode(vcs_repo.path))
 
         vcs_diff = vcs_repo.get_diff(
             commit1=target_commit, commit2=source_commit,
             ignore_whitespace=hide_whitespace_changes, context=diff_context)
         return vcs_diff
 
     def _is_merge_enabled(self, pull_request):
         return self._get_general_setting(
             pull_request, 'rhodecode_pr_merge_enabled')
 
     def _use_rebase_for_merging(self, pull_request):
         repo_type = pull_request.target_repo.repo_type
         if repo_type == 'hg':
             return self._get_general_setting(
                 pull_request, 'rhodecode_hg_use_rebase_for_merging')
         elif repo_type == 'git':
             return self._get_general_setting(
                 pull_request, 'rhodecode_git_use_rebase_for_merging')
 
         return False
 
     def _close_branch_before_merging(self, pull_request):
         repo_type = pull_request.target_repo.repo_type
         if repo_type == 'hg':
             return self._get_general_setting(
                 pull_request, 'rhodecode_hg_close_branch_before_merging')
         elif repo_type == 'git':
             return self._get_general_setting(
                 pull_request, 'rhodecode_git_close_branch_before_merging')
 
         return False
 
     def _get_general_setting(self, pull_request, settings_key, default=False):
         settings_model = VcsSettingsModel(repo=pull_request.target_repo)
         settings = settings_model.get_general_settings()
         return settings.get(settings_key, default)
 
     def _log_audit_action(self, action, action_data, user, pull_request):
         audit_logger.store(
             action=action,
             action_data=action_data,
             user=user,
             repo=pull_request.target_repo)
 
     def get_reviewer_functions(self):
         """
         Fetches functions for validation and fetching default reviewers.
         If available we use the EE package, else we fallback to CE
         package functions
         """
         try:
             from rc_reviewers.utils import get_default_reviewers_data
             from rc_reviewers.utils import validate_default_reviewers
         except ImportError:
             from rhodecode.apps.repository.utils import get_default_reviewers_data
             from rhodecode.apps.repository.utils import validate_default_reviewers
 
         return get_default_reviewers_data, validate_default_reviewers
 
 
 class MergeCheck(object):
     """
     Perform Merge Checks and returns a check object which stores information
     about merge errors, and merge conditions
     """
     TODO_CHECK = 'todo'
     PERM_CHECK = 'perm'
     REVIEW_CHECK = 'review'
     MERGE_CHECK = 'merge'
 
     def __init__(self):
         self.review_status = None
         self.merge_possible = None
         self.merge_msg = ''
         self.failed = None
         self.errors = []
         self.error_details = OrderedDict()
 
     def push_error(self, error_type, message, error_key, details):
         self.failed = True
         self.errors.append([error_type, message])
         self.error_details[error_key] = dict(
             details=details,
             error_type=error_type,
             message=message
         )
 
     @classmethod
     def validate(cls, pull_request, auth_user, translator, fail_early=False,
                  force_shadow_repo_refresh=False):
         _ = translator
         merge_check = cls()
 
         # permissions to merge
         user_allowed_to_merge = PullRequestModel().check_user_merge(
             pull_request, auth_user)
         if not user_allowed_to_merge:
             log.debug("MergeCheck: cannot merge, approval is pending.")
 
             msg = _('User `{}` not allowed to perform merge.').format(auth_user.username)
             merge_check.push_error('error', msg, cls.PERM_CHECK, auth_user.username)
             if fail_early:
                 return merge_check
 
         # permission to merge into the target branch
         target_commit_id = pull_request.target_ref_parts.commit_id
         if pull_request.target_ref_parts.type == 'branch':
             branch_name = pull_request.target_ref_parts.name
         else:
             # for mercurial we can always figure out the branch from the commit
             # in case of bookmark
             target_commit = pull_request.target_repo.get_commit(target_commit_id)
             branch_name = target_commit.branch
 
         rule, branch_perm = auth_user.get_rule_and_branch_permission(
             pull_request.target_repo.repo_name, branch_name)
         if branch_perm and branch_perm == 'branch.none':
             msg = _('Target branch `{}` changes rejected by rule {}.').format(
                 branch_name, rule)
             merge_check.push_error('error', msg, cls.PERM_CHECK, auth_user.username)
             if fail_early:
                 return merge_check
 
         # review status, must be always present
         review_status = pull_request.calculated_review_status()
         merge_check.review_status = review_status
 
         status_approved = review_status == ChangesetStatus.STATUS_APPROVED
         if not status_approved:
             log.debug("MergeCheck: cannot merge, approval is pending.")
 
             msg = _('Pull request reviewer approval is pending.')
 
             merge_check.push_error(
                 'warning', msg, cls.REVIEW_CHECK, review_status)
 
             if fail_early:
                 return merge_check
 
         # left over TODOs
         todos = CommentsModel().get_unresolved_todos(pull_request)
         if todos:
             log.debug("MergeCheck: cannot merge, {} "
                       "unresolved todos left.".format(len(todos)))
 
             if len(todos) == 1:
                 msg = _('Cannot merge, {} TODO still not resolved.').format(
                     len(todos))
             else:
                 msg = _('Cannot merge, {} TODOs still not resolved.').format(
                     len(todos))
 
             merge_check.push_error('warning', msg, cls.TODO_CHECK, todos)
 
             if fail_early:
                 return merge_check
 
         # merge possible, here is the filesystem simulation + shadow repo
         merge_status, msg = PullRequestModel().merge_status(
             pull_request, translator=translator,
             force_shadow_repo_refresh=force_shadow_repo_refresh)
         merge_check.merge_possible = merge_status
         merge_check.merge_msg = msg
         if not merge_status:
             log.debug(
                 "MergeCheck: cannot merge, pull request merge not possible.")
             merge_check.push_error('warning', msg, cls.MERGE_CHECK, None)
 
             if fail_early:
                 return merge_check
 
         log.debug('MergeCheck: is failed: %s', merge_check.failed)
         return merge_check
 
     @classmethod
     def get_merge_conditions(cls, pull_request, translator):
         _ = translator
         merge_details = {}
 
         model = PullRequestModel()
         use_rebase = model._use_rebase_for_merging(pull_request)
 
         if use_rebase:
             merge_details['merge_strategy'] = dict(
                 details={},
                 message=_('Merge strategy: rebase')
             )
         else:
             merge_details['merge_strategy'] = dict(
                 details={},
                 message=_('Merge strategy: explicit merge commit')
             )
 
         close_branch = model._close_branch_before_merging(pull_request)
         if close_branch:
             repo_type = pull_request.target_repo.repo_type
             if repo_type == 'hg':
                 close_msg = _('Source branch will be closed after merge.')
             elif repo_type == 'git':
                 close_msg = _('Source branch will be deleted after merge.')
 
             merge_details['close_branch'] = dict(
                 details={},
                 message=close_msg
             )
 
         return merge_details
 
 ChangeTuple = collections.namedtuple(
     'ChangeTuple', ['added', 'common', 'removed', 'total'])
 
 FileChangeTuple = collections.namedtuple(
     'FileChangeTuple', ['added', 'modified', 'removed'])