pytest: Use hardcoded login URLs in tests.

- It's not so easy to use pyramids URL generation methods from within the tests.
- Therfore we hardcode the URLs.

rhodecode/tests/__init__.py

@@ -1,254 +1,254 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import os
 import time
 import logging
 import datetime
 import hashlib
 import tempfile
 from os.path import join as jn
 
 from tempfile import _RandomNameSequence
 
 from paste.deploy import loadapp
 from paste.script.appinstall import SetupCommand
 
 import pylons
 import pylons.test
 from pylons import config, url
 from pylons.i18n.translation import _get_translator
 from pylons.util import ContextObj
 
 from routes.util import URLGenerator
 from webtest import TestApp
 from nose.plugins.skip import SkipTest
 import pytest
 
 from rhodecode import is_windows
+from rhodecode.config.routing import ADMIN_PREFIX
 from rhodecode.model.meta import Session
 from rhodecode.model.db import User
 from rhodecode.lib import auth
 from rhodecode.lib.helpers import flash, link_to
 from rhodecode.lib.utils2 import safe_unicode, safe_str
 
 # TODO: johbo: Solve time zone related issues and remove this tweak
 os.environ['TZ'] = 'UTC'
 if not is_windows:
     time.tzset()
 
 log = logging.getLogger(__name__)
 
 __all__ = [
     'get_new_dir', 'TestController', 'SkipTest',
     'url', 'link_to', 'ldap_lib_installed', 'clear_all_caches',
     'assert_session_flash', 'login_user',
     'TESTS_TMP_PATH', 'HG_REPO', 'GIT_REPO', 'SVN_REPO',
     'NEW_HG_REPO', 'NEW_GIT_REPO',
     'HG_FORK', 'GIT_FORK', 'TEST_USER_ADMIN_LOGIN', 'TEST_USER_ADMIN_PASS',
     'TEST_USER_REGULAR_LOGIN', 'TEST_USER_REGULAR_PASS',
     'TEST_USER_REGULAR_EMAIL', 'TEST_USER_REGULAR2_LOGIN',
     'TEST_USER_REGULAR2_PASS', 'TEST_USER_REGULAR2_EMAIL', 'TEST_HG_REPO',
     'TEST_HG_REPO_CLONE', 'TEST_HG_REPO_PULL', 'TEST_GIT_REPO',
     'TEST_GIT_REPO_CLONE', 'TEST_GIT_REPO_PULL', 'SCM_TESTS',
 ]
 
 # Invoke websetup with the current config file
 # SetupCommand('setup-app').run([config_file])
 
 # SOME GLOBALS FOR TESTS
 TEST_DIR = tempfile.gettempdir()
 
 TESTS_TMP_PATH = jn(TEST_DIR, 'rc_test_%s' % _RandomNameSequence().next())
 TEST_USER_ADMIN_LOGIN = 'test_admin'
 TEST_USER_ADMIN_PASS = 'test12'
 TEST_USER_ADMIN_EMAIL = '[e-mail]'
 
 TEST_USER_REGULAR_LOGIN = 'test_regular'
 TEST_USER_REGULAR_PASS = 'test12'
 TEST_USER_REGULAR_EMAIL = '[e-mail]'
 
 TEST_USER_REGULAR2_LOGIN = 'test_regular2'
 TEST_USER_REGULAR2_PASS = 'test12'
 TEST_USER_REGULAR2_EMAIL = '[e-mail]'
 
 HG_REPO = 'vcs_test_hg'
 GIT_REPO = 'vcs_test_git'
 SVN_REPO = 'vcs_test_svn'
 
 NEW_HG_REPO = 'vcs_test_hg_new'
 NEW_GIT_REPO = 'vcs_test_git_new'
 
 HG_FORK = 'vcs_test_hg_fork'
 GIT_FORK = 'vcs_test_git_fork'
 
 ## VCS
 SCM_TESTS = ['hg', 'git']
 uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))
 
 TEST_GIT_REPO = jn(TESTS_TMP_PATH, GIT_REPO)
 TEST_GIT_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)
 TEST_GIT_REPO_PULL = jn(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)
 
 TEST_HG_REPO = jn(TESTS_TMP_PATH, HG_REPO)
 TEST_HG_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)
 TEST_HG_REPO_PULL = jn(TESTS_TMP_PATH, 'vcshgpull%s' % uniq_suffix)
 
 TEST_REPO_PREFIX = 'vcs-test'
 
 
 # skip ldap tests if LDAP lib is not installed
 ldap_lib_installed = False
 try:
     import ldap
     ldap_lib_installed = True
 except ImportError:
     # means that python-ldap is not installed
     pass
 
 
 def clear_all_caches():
     from beaker.cache import cache_managers
     for _cache in cache_managers.values():
         _cache.clear()
 
 
 def get_new_dir(title):
     """
     Returns always new directory path.
     """
     from rhodecode.tests.vcs.utils import get_normalized_path
     name_parts = [TEST_REPO_PREFIX]
     if title:
         name_parts.append(title)
     hex_str = hashlib.sha1('%s %s' % (os.getpid(), time.time())).hexdigest()
     name_parts.append(hex_str)
     name = '-'.join(name_parts)
     path = os.path.join(TEST_DIR, name)
     return get_normalized_path(path)
 
 
 @pytest.mark.usefixtures('app', 'index_location')
 class TestController(object):
 
     maxDiff = None
 
     def log_user(self, username=TEST_USER_ADMIN_LOGIN,
                  password=TEST_USER_ADMIN_PASS):
         self._logged_username = username
         self._session = login_user_session(self.app, username, password)
         self.csrf_token = auth.get_csrf_token(self._session)
 
         return self._session['rhodecode_user']
 
     def logout_user(self):
         logout_user_session(self.app, auth.get_csrf_token(self._session))
         self.csrf_token = None
         self._logged_username = None
         self._session = None
 
     def _get_logged_user(self):
         return User.get_by_username(self._logged_username)
 
     # TODO: remove, use plain assert in tests
     def assertEqual(self, a, b, msg=None):
         if msg:
             assert a == b, msg
         else:
             assert a == b
 
 
 def login_user_session(
         app, username=TEST_USER_ADMIN_LOGIN, password=TEST_USER_ADMIN_PASS):
-    response = app.post(url(controller='login', action='index'),
-                        {'username': username,
-                         'password': password})
-
+    from rhodecode.tests.functional.test_login import login_url
+    response = app.post(
+        login_url,
+        {'username': username, 'password': password})
     if 'invalid user name' in response.body:
         pytest.fail('could not login using %s %s' % (username, password))
 
     assert response.status == '302 Found'
     ses = response.session['rhodecode_user']
     assert ses.get('username') == username
     response = response.follow()
     assert ses.get('is_authenticated')
 
     return response.session
 
 
 def logout_user_session(app, csrf_token):
-    app.post(
-        url(controller='login', action='logout'),
-        {'csrf_token': csrf_token}, status=302)
+    from rhodecode.tests.functional.test_login import logut_url
+    app.post(logut_url, {'csrf_token': csrf_token}, status=302)
 
 
 def login_user(app, username=TEST_USER_ADMIN_LOGIN,
                password=TEST_USER_ADMIN_PASS):
     return login_user_session(app, username, password)['rhodecode_user']
 
 
 def assert_session_flash(response=None, msg=None, category=None):
     """
     Assert on a flash message in the current session.
 
     :param msg: Required. The expected message. Will be evaluated if a
         :class:`LazyString` is passed in.
     :param response: Optional. For functional testing, pass in the response
         object. Otherwise don't pass in any value.
     :param category: Optional. If passed, the message category will be
         checked as well.
     """
     if msg is None:
         raise ValueError("Parameter msg is required.")
 
     messages = flash.pop_messages()
     message = messages[0]
 
     msg = _eval_if_lazy(msg)
     message_text = _eval_if_lazy(message.message)
 
     if msg not in message_text:
         msg = u'msg `%s` not found in session flash: got `%s` instead' % (
             msg, message_text)
         pytest.fail(safe_str(msg))
     if category:
         assert category == message.category
 
 
 def _eval_if_lazy(value):
     return value.eval() if hasattr(value, 'eval') else value
 
 
 def assert_not_in_session_flash(response, msg, category=None):
     assert 'flash' in response.session, 'Response session has no flash key'
     message_category, message_text = response.session['flash'][0]
     if msg in message_text:
         msg = u'msg `%s` found in session flash: got `%s` instead' % (
             msg, message_text)
         pytest.fail(safe_str(msg))
     if category:
         assert category == message_category
 
 
 def assert_session_flash_is_empty(response):
     if 'flash' in response.session:
         msg = 'flash messages are present in session:%s' % \
               response.session['flash'][0]
         pytest.fail(safe_str(msg))

rhodecode/tests/functional/test_admin_settings.py

@@ -1,621 +1,622 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import mock
 import pytest
 
 import rhodecode
+from rhodecode.config.routing import ADMIN_PREFIX
 from rhodecode.lib.utils2 import md5
 from rhodecode.model.db import RhodeCodeUi
 from rhodecode.model.meta import Session
 from rhodecode.model.settings import SettingsModel, IssueTrackerSettingsModel
 from rhodecode.tests import url, assert_session_flash
 from rhodecode.tests.utils import AssertResponse
 
 
 UPDATE_DATA_QUALNAME = (
     'rhodecode.controllers.admin.settings.SettingsController.get_update_data')
 
 
 @pytest.mark.usefixtures('autologin_user', 'app')
 class TestAdminSettingsController:
 
     @pytest.mark.parametrize('urlname', [
         'admin_settings_vcs',
         'admin_settings_mapping',
         'admin_settings_global',
         'admin_settings_visual',
         'admin_settings_email',
         'admin_settings_hooks',
         'admin_settings_search',
         'admin_settings_system',
     ])
     def test_simple_get(self, urlname, app):
         app.get(url(urlname))
 
     def test_create_custom_hook(self, csrf_token):
         response = self.app.post(
             url('admin_settings_hooks'),
             params={
                 'new_hook_ui_key': 'test_hooks_1',
                 'new_hook_ui_value': 'cd /tmp',
                 'csrf_token': csrf_token})
 
         response = response.follow()
         response.mustcontain('test_hooks_1')
         response.mustcontain('cd /tmp')
 
     def test_create_custom_hook_delete(self, csrf_token):
         response = self.app.post(
             url('admin_settings_hooks'),
             params={
                 'new_hook_ui_key': 'test_hooks_2',
                 'new_hook_ui_value': 'cd /tmp2',
                 'csrf_token': csrf_token})
 
         response = response.follow()
         response.mustcontain('test_hooks_2')
         response.mustcontain('cd /tmp2')
 
         hook_id = SettingsModel().get_ui_by_key('test_hooks_2').ui_id
 
         # delete
         self.app.post(
             url('admin_settings_hooks'),
             params={'hook_id': hook_id, 'csrf_token': csrf_token})
         response = self.app.get(url('admin_settings_hooks'))
         response.mustcontain(no=['test_hooks_2'])
         response.mustcontain(no=['cd /tmp2'])
 
     def test_system_update_new_version(self):
         update_data = {
             'versions': [
                 {
                     'version': '100.3.1415926535',
                     'general': 'The latest version we are ever going to ship'
                 },
                 {
                     'version': '0.0.0',
                     'general': 'The first version we ever shipped'
                 }
             ]
         }
         with mock.patch(UPDATE_DATA_QUALNAME, return_value=update_data):
             response = self.app.get(url('admin_settings_system_update'))
             response.mustcontain('A <b>new version</b> is available')
 
     def test_system_update_nothing_new(self):
         update_data = {
             'versions': [
                 {
                     'version': '0.0.0',
                     'general': 'The first version we ever shipped'
                 }
             ]
         }
         with mock.patch(UPDATE_DATA_QUALNAME, return_value=update_data):
             response = self.app.get(url('admin_settings_system_update'))
             response.mustcontain(
                 'You already have the <b>latest</b> stable version.')
 
     def test_system_update_bad_response(self):
         with mock.patch(UPDATE_DATA_QUALNAME, side_effect=ValueError('foo')):
             response = self.app.get(url('admin_settings_system_update'))
             response.mustcontain(
                 'Bad data sent from update server')
 
 
 @pytest.mark.usefixtures('autologin_user', 'app')
 class TestAdminSettingsGlobal:
 
     def test_pre_post_code_code_active(self, csrf_token):
         pre_code = 'rc-pre-code-187652122'
         post_code = 'rc-postcode-98165231'
 
         response = self.post_and_verify_settings({
             'rhodecode_pre_code': pre_code,
             'rhodecode_post_code': post_code,
             'csrf_token': csrf_token,
         })
 
         response = response.follow()
         response.mustcontain(pre_code, post_code)
 
     def test_pre_post_code_code_inactive(self, csrf_token):
         pre_code = 'rc-pre-code-187652122'
         post_code = 'rc-postcode-98165231'
         response = self.post_and_verify_settings({
             'rhodecode_pre_code': '',
             'rhodecode_post_code': '',
             'csrf_token': csrf_token,
         })
 
         response = response.follow()
         response.mustcontain(no=[pre_code, post_code])
 
     def test_captcha_activate(self, csrf_token):
         self.post_and_verify_settings({
             'rhodecode_captcha_private_key': '1234567890',
             'rhodecode_captcha_public_key': '1234567890',
             'csrf_token': csrf_token,
         })
 
-        response = self.app.get(url('register'))
+        response = self.app.get(ADMIN_PREFIX + '/register')
         response.mustcontain('captcha')
 
     def test_captcha_deactivate(self, csrf_token):
         self.post_and_verify_settings({
             'rhodecode_captcha_private_key': '',
             'rhodecode_captcha_public_key': '1234567890',
             'csrf_token': csrf_token,
         })
 
-        response = self.app.get(url('register'))
+        response = self.app.get(ADMIN_PREFIX + '/register')
         response.mustcontain(no=['captcha'])
 
     def test_title_change(self, csrf_token):
         old_title = 'RhodeCode'
         new_title = old_title + '_changed'
 
         for new_title in ['Changed', 'Żółwik', old_title]:
             response = self.post_and_verify_settings({
                 'rhodecode_title': new_title,
                 'csrf_token': csrf_token,
             })
 
             response = response.follow()
             response.mustcontain(
                 """<div class="branding">- %s</div>""" % new_title)
 
     def post_and_verify_settings(self, settings):
         old_title = 'RhodeCode'
         old_realm = 'RhodeCode authentication'
         params = {
             'rhodecode_title': old_title,
             'rhodecode_realm': old_realm,
             'rhodecode_pre_code': '',
             'rhodecode_post_code': '',
             'rhodecode_captcha_private_key': '',
             'rhodecode_captcha_public_key': '',
         }
         params.update(settings)
         response = self.app.post(url('admin_settings_global'), params=params)
 
         assert_session_flash(response, 'Updated application settings')
         app_settings = SettingsModel().get_all_settings()
         del settings['csrf_token']
         for key, value in settings.iteritems():
             assert app_settings[key] == value.decode('utf-8')
 
         return response
 
 
 @pytest.mark.usefixtures('autologin_user', 'app')
 class TestAdminSettingsVcs:
 
     def test_contains_svn_default_patterns(self, app):
         response = app.get(url('admin_settings_vcs'))
         expected_patterns = [
             '/trunk',
             '/branches/*',
             '/tags/*',
         ]
         for pattern in expected_patterns:
             response.mustcontain(pattern)
 
     def test_add_new_svn_branch_and_tag_pattern(
             self, app, backend_svn, form_defaults, disable_sql_cache,
             csrf_token):
         form_defaults.update({
             'new_svn_branch': '/exp/branches/*',
             'new_svn_tag': '/important_tags/*',
             'csrf_token': csrf_token,
         })
 
         response = app.post(
             url('admin_settings_vcs'), params=form_defaults, status=302)
         response = response.follow()
 
         # Expect to find the new values on the page
         response.mustcontain('/exp/branches/*')
         response.mustcontain('/important_tags/*')
 
         # Expect that those patterns are used to match branches and tags now
         repo = backend_svn['svn-simple-layout'].scm_instance()
         assert 'exp/branches/exp-sphinx-docs' in repo.branches
         assert 'important_tags/v0.5' in repo.tags
 
     def test_add_same_svn_value_twice_shows_an_error_message(
             self, app, form_defaults, csrf_token, settings_util):
         settings_util.create_rhodecode_ui('vcs_svn_branch', '/test')
         settings_util.create_rhodecode_ui('vcs_svn_tag', '/test')
 
         response = app.post(
             url('admin_settings_vcs'),
             params={
                 'paths_root_path': form_defaults['paths_root_path'],
                 'new_svn_branch': '/test',
                 'new_svn_tag': '/test',
                 'csrf_token': csrf_token,
             },
             status=200)
 
         response.mustcontain("Pattern already exists")
         response.mustcontain("Some form inputs contain invalid data.")
 
     @pytest.mark.parametrize('section', [
         'vcs_svn_branch',
         'vcs_svn_tag',
     ])
     def test_delete_svn_patterns(
             self, section, app, csrf_token, settings_util):
         setting = settings_util.create_rhodecode_ui(
             section, '/test_delete', cleanup=False)
 
         app.post(
             url('admin_settings_vcs'),
             params={
                 '_method': 'delete',
                 'delete_svn_pattern': setting.ui_id,
                 'csrf_token': csrf_token},
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest'})
 
     @pytest.mark.parametrize('section', [
         'vcs_svn_branch',
         'vcs_svn_tag',
     ])
     def test_delete_svn_patterns_raises_400_when_no_xhr(
             self, section, app, csrf_token, settings_util):
         setting = settings_util.create_rhodecode_ui(section, '/test_delete')
 
         app.post(
             url('admin_settings_vcs'),
             params={
                 '_method': 'delete',
                 'delete_svn_pattern': setting.ui_id,
                 'csrf_token': csrf_token},
             status=400)
 
     def test_extensions_hgsubversion(self, app, form_defaults, csrf_token):
         form_defaults.update({
             'csrf_token': csrf_token,
             'extensions_hgsubversion': 'True',
         })
         response = app.post(
             url('admin_settings_vcs'),
             params=form_defaults,
             status=302)
 
         response = response.follow()
         extensions_input = (
             '<input id="extensions_hgsubversion" '
             'name="extensions_hgsubversion" type="checkbox" '
             'value="True" checked="checked" />')
         response.mustcontain(extensions_input)
 
     def test_has_a_section_for_pull_request_settings(self, app):
         response = app.get(url('admin_settings_vcs'))
         response.mustcontain('Pull Request Settings')
 
     def test_has_an_input_for_invalidation_of_inline_comments(
             self, app):
         response = app.get(url('admin_settings_vcs'))
         assert_response = AssertResponse(response)
         assert_response.one_element_exists(
             '[name=rhodecode_use_outdated_comments]')
 
     @pytest.mark.parametrize('new_value', [True, False])
     def test_allows_to_change_invalidation_of_inline_comments(
             self, app, form_defaults, csrf_token, new_value):
         setting_key = 'use_outdated_comments'
         setting = SettingsModel().create_or_update_setting(
             setting_key, not new_value, 'bool')
         Session().add(setting)
         Session().commit()
 
         form_defaults.update({
             'csrf_token': csrf_token,
             'rhodecode_use_outdated_comments': str(new_value),
         })
         response = app.post(
             url('admin_settings_vcs'),
             params=form_defaults,
             status=302)
         response = response.follow()
         setting = SettingsModel().get_setting_by_name(setting_key)
         assert setting.app_settings_value is new_value
 
     @pytest.fixture
     def disable_sql_cache(self, request):
         patcher = mock.patch(
             'rhodecode.lib.caching_query.FromCache.process_query')
         request.addfinalizer(patcher.stop)
         patcher.start()
 
     @pytest.fixture
     def form_defaults(self):
         from rhodecode.controllers.admin.settings import SettingsController
         controller = SettingsController()
         return controller._form_defaults()
 
     # TODO: johbo: What we really want is to checkpoint before a test run and
     # reset the session afterwards.
     @pytest.fixture(scope='class', autouse=True)
     def cleanup_settings(self, request, pylonsapp):
         ui_id = RhodeCodeUi.ui_id
         original_ids = list(
             r.ui_id for r in RhodeCodeUi.query().values(ui_id))
 
         @request.addfinalizer
         def cleanup():
             RhodeCodeUi.query().filter(
                 ui_id.notin_(original_ids)).delete(False)
 
 
 @pytest.mark.usefixtures('autologin_user', 'app')
 class TestLabsSettings(object):
     def test_get_settings_page_disabled(self):
         with mock.patch.dict(rhodecode.CONFIG,
                              {'labs_settings_active': 'false'}):
             response = self.app.get(url('admin_settings_labs'), status=302)
 
         assert response.location.endswith(url('admin_settings'))
 
     def test_get_settings_page_enabled(self):
         from rhodecode.controllers.admin import settings
         lab_settings = [
             settings.LabSetting(
                 key='rhodecode_bool',
                 type='bool',
                 group='bool group',
                 label='bool label',
                 help='bool help'
             ),
             settings.LabSetting(
                 key='rhodecode_text',
                 type='unicode',
                 group='text group',
                 label='text label',
                 help='text help'
             ),
         ]
         with mock.patch.dict(rhodecode.CONFIG,
                              {'labs_settings_active': 'true'}):
             with mock.patch.object(settings, '_LAB_SETTINGS', lab_settings):
                 response = self.app.get(url('admin_settings_labs'))
 
         assert '<label>bool group:</label>' in response
         assert '<label for="rhodecode_bool">bool label</label>' in response
         assert '<p class="help-block">bool help</p>' in response
         assert 'name="rhodecode_bool" type="checkbox"' in response
 
         assert '<label>text group:</label>' in response
         assert '<label for="rhodecode_text">text label</label>' in response
         assert '<p class="help-block">text help</p>' in response
         assert 'name="rhodecode_text" size="60" type="text"' in response
 
     @pytest.mark.parametrize('setting_name', [
         'proxy_subversion_http_requests',
         'hg_use_rebase_for_merging',
     ])
     def test_update_boolean_settings(self, csrf_token, setting_name):
         self.app.post(
             url('admin_settings_labs'),
             params={
                 'rhodecode_{}'.format(setting_name): 'true',
                 'csrf_token': csrf_token,
             })
         setting = SettingsModel().get_setting_by_name(setting_name)
         assert setting.app_settings_value
 
         self.app.post(
             url('admin_settings_labs'),
             params={
                 'rhodecode_{}'.format(setting_name): 'false',
                 'csrf_token': csrf_token,
             })
         setting = SettingsModel().get_setting_by_name(setting_name)
         assert not setting.app_settings_value
 
     @pytest.mark.parametrize('setting_name', [
         'subversion_http_server_url',
     ])
     def test_update_string_settings(self, csrf_token, setting_name):
         self.app.post(
             url('admin_settings_labs'),
             params={
                 'rhodecode_{}'.format(setting_name): 'Test 1',
                 'csrf_token': csrf_token,
             })
         setting = SettingsModel().get_setting_by_name(setting_name)
         assert setting.app_settings_value == 'Test 1'
 
         self.app.post(
             url('admin_settings_labs'),
             params={
                 'rhodecode_{}'.format(setting_name): ' Test 2 ',
                 'csrf_token': csrf_token,
             })
         setting = SettingsModel().get_setting_by_name(setting_name)
         assert setting.app_settings_value == 'Test 2'
 
 
 @pytest.mark.usefixtures('app')
 class TestOpenSourceLicenses(object):
     def test_records_are_displayed(self, autologin_user):
         sample_licenses = {
             "python2.7-pytest-2.7.1": {
                 "UNKNOWN": None
             },
             "python2.7-Markdown-2.6.2": {
                 "BSD-3-Clause": "http://spdx.org/licenses/BSD-3-Clause"
             }
         }
         read_licenses_patch = mock.patch(
             'rhodecode.controllers.admin.settings.read_opensource_licenses',
             return_value=sample_licenses)
         with read_licenses_patch:
             response = self.app.get(
                 url('admin_settings_open_source'), status=200)
 
         assert_response = AssertResponse(response)
         assert_response.element_contains(
             '.panel-heading', 'Licenses of Third Party Packages')
         for name in sample_licenses:
             response.mustcontain(name)
             for license in sample_licenses[name]:
                 assert_response.element_contains('.panel-body', license)
 
     def test_records_can_be_read(self, autologin_user):
         response = self.app.get(url('admin_settings_open_source'), status=200)
         assert_response = AssertResponse(response)
         assert_response.element_contains(
             '.panel-heading', 'Licenses of Third Party Packages')
 
     def test_forbidden_when_normal_user(self, autologin_regular_user):
         self.app.get(
             url('admin_settings_open_source'), status=403)
 
 
 @pytest.mark.usefixtures("app")
 class TestAdminSettingsIssueTracker:
     RC_PREFIX = 'rhodecode_'
     SHORT_PATTERN_KEY = 'issuetracker_pat_'
     PATTERN_KEY = RC_PREFIX + SHORT_PATTERN_KEY
 
     def test_issuetracker_index(self, autologin_user):
         response = self.app.get(url('admin_settings_issuetracker'))
         assert response.status_code == 200
 
     def test_add_issuetracker_pattern(
             self, request, autologin_user, csrf_token):
         pattern = 'issuetracker_pat'
         another_pattern = pattern+'1'
         post_url = url('admin_settings_issuetracker_save')
         post_data = {
             'new_pattern_pattern_0': pattern,
             'new_pattern_url_0': 'url',
             'new_pattern_prefix_0': 'prefix',
             'new_pattern_description_0': 'description',
             'new_pattern_pattern_1': another_pattern,
             'new_pattern_url_1': 'url1',
             'new_pattern_prefix_1': 'prefix1',
             'new_pattern_description_1': 'description1',
             'csrf_token': csrf_token
         }
         self.app.post(post_url, post_data, status=302)
         settings = SettingsModel().get_all_settings()
         self.uid = md5(pattern)
         assert settings[self.PATTERN_KEY+self.uid] == pattern
         self.another_uid = md5(another_pattern)
         assert settings[self.PATTERN_KEY+self.another_uid] == another_pattern
 
         @request.addfinalizer
         def cleanup():
             defaults = SettingsModel().get_all_settings()
 
             entries = [name for name in defaults if (
                 (self.uid in name) or (self.another_uid) in name)]
             start = len(self.RC_PREFIX)
             for del_key in entries:
                 # TODO: anderson: get_by_name needs name without prefix
                 entry = SettingsModel().get_setting_by_name(del_key[start:])
                 Session().delete(entry)
 
             Session().commit()
 
     def test_edit_issuetracker_pattern(
             self, autologin_user, backend, csrf_token, request):
         old_pattern = 'issuetracker_pat'
         old_uid = md5(old_pattern)
         pattern = 'issuetracker_pat_new'
         self.new_uid = md5(pattern)
 
         SettingsModel().create_or_update_setting(
             self.SHORT_PATTERN_KEY+old_uid, old_pattern, 'unicode')
 
         post_url = url('admin_settings_issuetracker_save')
         post_data = {
             'new_pattern_pattern_0': pattern,
             'new_pattern_url_0': 'url',
             'new_pattern_prefix_0': 'prefix',
             'new_pattern_description_0': 'description',
             'uid': old_uid,
             'csrf_token': csrf_token
         }
         self.app.post(post_url, post_data, status=302)
         settings = SettingsModel().get_all_settings()
         assert settings[self.PATTERN_KEY+self.new_uid] == pattern
         assert self.PATTERN_KEY+old_uid not in settings
 
         @request.addfinalizer
         def cleanup():
             IssueTrackerSettingsModel().delete_entries(self.new_uid)
 
     def test_replace_issuetracker_pattern_description(
             self, autologin_user, csrf_token, request, settings_util):
         prefix = 'issuetracker'
         pattern = 'issuetracker_pat'
         self.uid = md5(pattern)
         pattern_key = '_'.join([prefix, 'pat', self.uid])
         rc_pattern_key = '_'.join(['rhodecode', pattern_key])
         desc_key = '_'.join([prefix, 'desc', self.uid])
         rc_desc_key = '_'.join(['rhodecode', desc_key])
         new_description = 'new_description'
 
         settings_util.create_rhodecode_setting(
             pattern_key, pattern, 'unicode', cleanup=False)
         settings_util.create_rhodecode_setting(
             desc_key, 'old description', 'unicode', cleanup=False)
 
         post_url = url('admin_settings_issuetracker_save')
         post_data = {
             'new_pattern_pattern_0': pattern,
             'new_pattern_url_0': 'url',
             'new_pattern_prefix_0': 'prefix',
             'new_pattern_description_0': new_description,
             'uid': self.uid,
             'csrf_token': csrf_token
         }
         self.app.post(post_url, post_data, status=302)
         settings = SettingsModel().get_all_settings()
         assert settings[rc_pattern_key] == pattern
         assert settings[rc_desc_key] == new_description
 
         @request.addfinalizer
         def cleanup():
             IssueTrackerSettingsModel().delete_entries(self.uid)
 
     def test_delete_issuetracker_pattern(
             self, autologin_user, backend, csrf_token, settings_util):
         pattern = 'issuetracker_pat'
         uid = md5(pattern)
         settings_util.create_rhodecode_setting(
             self.SHORT_PATTERN_KEY+uid, pattern, 'unicode', cleanup=False)
 
         post_url = url('admin_issuetracker_delete')
         post_data = {
             '_method': 'delete',
             'uid': uid,
             'csrf_token': csrf_token
         }
         self.app.post(post_url, post_data, status=302)
         settings = SettingsModel().get_all_settings()
         assert 'rhodecode_%s%s' % (self.SHORT_PATTERN_KEY, uid) not in settings

rhodecode/tests/functional/test_login.py

@@ -1,511 +1,519 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  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 urlparse
 
 import mock
 import pytest
 
+from rhodecode.config.routing import ADMIN_PREFIX
 from rhodecode.tests import (
     assert_session_flash, url, HG_REPO, TEST_USER_ADMIN_LOGIN)
 from rhodecode.tests.fixture import Fixture
+from rhodecode.tests.utils import AssertResponse, get_session_from_response
 from rhodecode.lib.auth import check_password, generate_auth_token
 from rhodecode.lib import helpers as h
 from rhodecode.model.auth_token import AuthTokenModel
 from rhodecode.model import validators
 from rhodecode.model.db import User, Notification
 from rhodecode.model.meta import Session
 
 fixture = Fixture()
 
+# Hardcode URLs because we don't have a request object to use
+# pyramids URL generation methods.
+login_url = ADMIN_PREFIX + '/login'
+logut_url = ADMIN_PREFIX + '/logout'
+register_url = ADMIN_PREFIX + '/register'
+pwd_reset_url = ADMIN_PREFIX + '/password_reset'
+pwd_reset_confirm_url = ADMIN_PREFIX + '/password_reset_confirmation'
+
 
 @pytest.mark.usefixtures('app')
 class TestLoginController:
     destroy_users = set()
 
     @classmethod
     def teardown_class(cls):
         fixture.destroy_users(cls.destroy_users)
 
     def teardown_method(self, method):
         for n in Notification.query().all():
             Session().delete(n)
 
         Session().commit()
         assert Notification.query().all() == []
 
     def test_index(self):
-        response = self.app.get(url(controller='login', action='index'))
+        response = self.app.get(login_url)
         assert response.status == '200 OK'
         # Test response...
 
     def test_login_admin_ok(self):
-        response = self.app.post(url(controller='login', action='index'),
+        response = self.app.post(login_url,
                                  {'username': 'test_admin',
                                   'password': 'test12'})
         assert response.status == '302 Found'
-        username = response.session['rhodecode_user'].get('username')
+        session = get_session_from_response(response)
+        username = session['rhodecode_user'].get('username')
         assert username == 'test_admin'
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
 
     def test_login_regular_ok(self):
-        response = self.app.post(url(controller='login', action='index'),
+        response = self.app.post(login_url,
                                  {'username': 'test_regular',
                                   'password': 'test12'})
 
         assert response.status == '302 Found'
-        username = response.session['rhodecode_user'].get('username')
+        session = get_session_from_response(response)
+        username = session['rhodecode_user'].get('username')
         assert username == 'test_regular'
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
 
     def test_login_ok_came_from(self):
         test_came_from = '/_admin/users?branch=stable'
-        response = self.app.post(url(controller='login', action='index',
-                                     came_from=test_came_from),
-                                 {'username': 'test_admin',
-                                  'password': 'test12'})
+        _url = '{}?came_from={}'.format(login_url, test_came_from)
+        response = self.app.post(
+            _url, {'username': 'test_admin', 'password': 'test12'})
         assert response.status == '302 Found'
         assert 'branch=stable' in response.location
         response = response.follow()
 
         assert response.status == '200 OK'
         response.mustcontain('Users administration')
 
     def test_redirect_to_login_with_get_args(self):
         with fixture.anon_access(False):
             kwargs = {'branch': 'stable'}
             response = self.app.get(
                 url('summary_home', repo_name=HG_REPO, **kwargs))
             assert response.status == '302 Found'
             response_query = urlparse.parse_qsl(response.location)
             assert 'branch=stable' in response_query[0][1]
 
     def test_login_form_with_get_args(self):
-        kwargs = {'branch': 'stable'}
-        response = self.app.get(
-            url(controller='login', action='index',
-                came_from='/_admin/users', **kwargs))
-        assert 'branch=stable' in response.form.action
+        _url = '{}?came_from=/_admin/users,branch=stable'.format(login_url)
+        response = self.app.get(_url)
+        assert 'branch%3Dstable' in response.form.action
 
     @pytest.mark.parametrize("url_came_from", [
-        ('data:text/html,<script>window.alert("xss")</script>',),
-        ('mailto:[e-mail]',),
-        ('file:///etc/passwd',),
-        ('ftp://some.ftp.server',),
-        ('http://other.domain',),
-        ('/\r\nX-Forwarded-Host: http://example.org',),
+        'data:text/html,<script>window.alert("xss")</script>',
+        'mailto:[e-mail]',
+        'file:///etc/passwd',
+        'ftp://some.ftp.server',
+        'http://other.domain',
+        '/\r\nX-Forwarded-Host: http://example.org',
     ])
     def test_login_bad_came_froms(self, url_came_from):
-        response = self.app.post(url(controller='login', action='index',
-                                     came_from=url_came_from),
-                                 {'username': 'test_admin',
-                                  'password': 'test12'})
+        _url = '{}?came_from={}'.format(login_url, url_came_from)
+        response = self.app.post(
+            _url,
+            {'username': 'test_admin', 'password': 'test12'})
         assert response.status == '302 Found'
-        assert response.tmpl_context.came_from == '/'
-
         response = response.follow()
         assert response.status == '200 OK'
+        assert response.request.path == '/'
 
     def test_login_short_password(self):
-        response = self.app.post(url(controller='login', action='index'),
+        response = self.app.post(login_url,
                                  {'username': 'test_admin',
                                   'password': 'as'})
         assert response.status == '200 OK'
 
         response.mustcontain('Enter 3 characters or more')
 
     def test_login_wrong_non_ascii_password(self, user_regular):
         response = self.app.post(
-            url(controller='login', action='index'),
+            login_url,
             {'username': user_regular.username,
              'password': u'invalid-non-asci\xe4'.encode('utf8')})
 
         response.mustcontain('invalid user name')
         response.mustcontain('invalid password')
 
     def test_login_with_non_ascii_password(self, user_util):
         password = u'valid-non-ascii\xe4'
         user = user_util.create_user(password=password)
         response = self.app.post(
-            url(controller='login', action='index'),
+            login_url,
             {'username': user.username,
              'password': password.encode('utf-8')})
         assert response.status_code == 302
 
     def test_login_wrong_username_password(self):
-        response = self.app.post(url(controller='login', action='index'),
+        response = self.app.post(login_url,
                                  {'username': 'error',
                                   'password': 'test12'})
 
         response.mustcontain('invalid user name')
         response.mustcontain('invalid password')
 
     def test_login_admin_ok_password_migration(self, real_crypto_backend):
         from rhodecode.lib import auth
 
         # create new user, with sha256 password
         temp_user = 'test_admin_sha256'
         user = fixture.create_user(temp_user)
         user.password = auth._RhodeCodeCryptoSha256().hash_create(
             b'test123')
         Session().add(user)
         Session().commit()
         self.destroy_users.add(temp_user)
-        response = self.app.post(url(controller='login', action='index'),
+        response = self.app.post(login_url,
                                  {'username': temp_user,
                                   'password': 'test123'})
 
         assert response.status == '302 Found'
-        username = response.session['rhodecode_user'].get('username')
+        session = get_session_from_response(response)
+        username = session['rhodecode_user'].get('username')
         assert username == temp_user
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
 
         # new password should be bcrypted, after log-in and transfer
         user = User.get_by_username(temp_user)
         assert user.password.startswith('$')
 
     # REGISTRATIONS
     def test_register(self):
-        response = self.app.get(url(controller='login', action='register'))
+        response = self.app.get(register_url)
         response.mustcontain('Create an Account')
 
     def test_register_err_same_username(self):
         uname = 'test_admin'
         response = self.app.post(
-            url(controller='login', action='register'),
+            register_url,
             {
                 'username': uname,
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
+        assertr = AssertResponse(response)
         msg = validators.ValidUsername()._messages['username_exists']
-        msg = h.html_escape(msg % {'username': uname})
-        response.mustcontain(msg)
+        msg = msg % {'username': uname}
+        assertr.element_contains('#username+.error-message', msg)
 
     def test_register_err_same_email(self):
         response = self.app.post(
-            url(controller='login', action='register'),
+            register_url,
             {
                 'username': 'test_admin_0',
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
+        assertr = AssertResponse(response)
         msg = validators.UniqSystemEmail()()._messages['email_taken']
-        response.mustcontain(msg)
+        assertr.element_contains('#email+.error-message', msg)
 
     def test_register_err_same_email_case_sensitive(self):
         response = self.app.post(
-            url(controller='login', action='register'),
+            register_url,
             {
                 'username': 'test_admin_1',
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
+        assertr = AssertResponse(response)
         msg = validators.UniqSystemEmail()()._messages['email_taken']
-        response.mustcontain(msg)
+        assertr.element_contains('#email+.error-message', msg)
 
     def test_register_err_wrong_data(self):
         response = self.app.post(
-            url(controller='login', action='register'),
+            register_url,
             {
                 'username': 'xs',
                 'password': 'test',
                 'password_confirmation': 'test',
                 'email': 'goodmailm',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
         assert response.status == '200 OK'
         response.mustcontain('An email address must contain a single @')
         response.mustcontain('Enter a value 6 characters long or more')
 
     def test_register_err_username(self):
         response = self.app.post(
-            url(controller='login', action='register'),
+            register_url,
             {
                 'username': 'error user',
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': 'goodmailm',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         response.mustcontain('An email address must contain a single @')
         response.mustcontain(
             'Username may only contain '
             'alphanumeric characters underscores, '
             'periods or dashes and must begin with '
             'alphanumeric character')
 
     def test_register_err_case_sensitive(self):
         usr = 'Test_Admin'
         response = self.app.post(
-            url(controller='login', action='register'),
+            register_url,
             {
                 'username': usr,
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': 'goodmailm',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
-        response.mustcontain('An email address must contain a single @')
+        assertr = AssertResponse(response)
         msg = validators.ValidUsername()._messages['username_exists']
-        msg = h.html_escape(msg % {'username': usr})
-        response.mustcontain(msg)
+        msg = msg % {'username': usr}
+        assertr.element_contains('#username+.error-message', msg)
 
     def test_register_special_chars(self):
         response = self.app.post(
-            url(controller='login', action='register'),
+            register_url,
             {
                 'username': 'xxxaxn',
                 'password': 'ąćźżąśśśś',
                 'password_confirmation': 'ąćźżąśśśś',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         msg = validators.ValidPassword()._messages['invalid_password']
         response.mustcontain(msg)
 
     def test_register_password_mismatch(self):
         response = self.app.post(
-            url(controller='login', action='register'),
+            register_url,
             {
                 'username': 'xs',
                 'password': '123qwe',
                 'password_confirmation': 'qwe123',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
         msg = validators.ValidPasswordsMatch()._messages['password_mismatch']
         response.mustcontain(msg)
 
     def test_register_ok(self):
         username = 'test_regular4'
         password = 'qweqwe'
         email = '[e-mail]'
         name = 'testname'
         lastname = 'testlastname'
 
         response = self.app.post(
-            url(controller='login', action='register'),
+            register_url,
             {
                 'username': username,
                 'password': password,
                 'password_confirmation': password,
                 'email': email,
                 'firstname': name,
                 'lastname': lastname,
                 'admin': True
             }
         )  # This should be overriden
         assert response.status == '302 Found'
         assert_session_flash(
             response, 'You have successfully registered with RhodeCode')
 
         ret = Session().query(User).filter(
             User.username == 'test_regular4').one()
         assert ret.username == username
         assert check_password(password, ret.password)
         assert ret.email == email
         assert ret.name == name
         assert ret.lastname == lastname
         assert ret.api_key is not None
         assert not ret.admin
 
     def test_forgot_password_wrong_mail(self):
         bad_email = '[e-mail]'
         response = self.app.post(
-            url(controller='login', action='password_reset'),
+            pwd_reset_url,
             {'email': bad_email, }
         )
 
         msg = validators.ValidSystemEmail()._messages['non_existing_email']
         msg = h.html_escape(msg % {'email': bad_email})
         response.mustcontain()
 
     def test_forgot_password(self):
-        response = self.app.get(url(controller='login',
-                                    action='password_reset'))
+        response = self.app.get(pwd_reset_url)
         assert response.status == '200 OK'
 
         username = 'test_password_reset_1'
         password = 'qweqwe'
         email = '[e-mail]'
         name = 'passwd'
         lastname = 'reset'
 
         new = User()
         new.username = username
         new.password = password
         new.email = email
         new.name = name
         new.lastname = lastname
         new.api_key = generate_auth_token(username)
         Session().add(new)
         Session().commit()
 
-        response = self.app.post(url(controller='login',
-                                     action='password_reset'),
+        response = self.app.post(pwd_reset_url,
                                  {'email': email, })
 
         assert_session_flash(
             response, 'Your password reset link was sent')
 
         response = response.follow()
 
         # BAD KEY
 
         key = "bad"
-        response = self.app.get(url(controller='login',
-                                    action='password_reset_confirmation',
-                                    key=key))
+        confirm_url = '{}?key={}'.format(pwd_reset_confirm_url, key)
+        response = self.app.get(confirm_url)
         assert response.status == '302 Found'
-        assert response.location.endswith(url('reset_password'))
+        assert response.location.endswith(pwd_reset_url)
 
         # GOOD KEY
 
         key = User.get_by_username(username).api_key
-        response = self.app.get(url(controller='login',
-                                    action='password_reset_confirmation',
-                                    key=key))
+        confirm_url = '{}?key={}'.format(pwd_reset_confirm_url, key)
+        response = self.app.get(confirm_url)
         assert response.status == '302 Found'
-        assert response.location.endswith(url('login_home'))
+        assert response.location.endswith(login_url)
 
         assert_session_flash(
             response,
             'Your password reset was successful, '
             'a new password has been sent to your email')
 
         response = response.follow()
 
     def _get_api_whitelist(self, values=None):
         config = {'api_access_controllers_whitelist': values or []}
         return config
 
     @pytest.mark.parametrize("test_name, auth_token", [
         ('none', None),
         ('empty_string', ''),
         ('fake_number', '123456'),
         ('proper_auth_token', None)
     ])
     def test_access_not_whitelisted_page_via_auth_token(self, test_name,
                                                         auth_token):
         whitelist = self._get_api_whitelist([])
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
             assert [] == whitelist['api_access_controllers_whitelist']
             if test_name == 'proper_auth_token':
                 # use builtin if api_key is None
                 auth_token = User.get_first_admin().api_key
 
             with fixture.anon_access(False):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  repo_name=HG_REPO, revision='tip',
                                  api_key=auth_token),
                              status=302)
 
     @pytest.mark.parametrize("test_name, auth_token, code", [
         ('none', None, 302),
         ('empty_string', '', 302),
         ('fake_number', '123456', 302),
         ('proper_auth_token', None, 200)
     ])
     def test_access_whitelisted_page_via_auth_token(self, test_name,
                                                     auth_token, code):
         whitelist = self._get_api_whitelist(
             ['ChangesetController:changeset_raw'])
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
             assert ['ChangesetController:changeset_raw'] == \
                 whitelist['api_access_controllers_whitelist']
             if test_name == 'proper_auth_token':
                 auth_token = User.get_first_admin().api_key
 
             with fixture.anon_access(False):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  repo_name=HG_REPO, revision='tip',
                                  api_key=auth_token),
                              status=code)
 
     def test_access_page_via_extra_auth_token(self):
         whitelist = self._get_api_whitelist(
             ['ChangesetController:changeset_raw'])
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
             assert ['ChangesetController:changeset_raw'] == \
                 whitelist['api_access_controllers_whitelist']
 
             new_auth_token = AuthTokenModel().create(
                 TEST_USER_ADMIN_LOGIN, 'test')
             Session().commit()
             with fixture.anon_access(False):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  repo_name=HG_REPO, revision='tip',
                                  api_key=new_auth_token.api_key),
                              status=200)
 
     def test_access_page_via_expired_auth_token(self):
         whitelist = self._get_api_whitelist(
             ['ChangesetController:changeset_raw'])
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
             assert ['ChangesetController:changeset_raw'] == \
                 whitelist['api_access_controllers_whitelist']
 
             new_auth_token = AuthTokenModel().create(
                 TEST_USER_ADMIN_LOGIN, 'test')
             Session().commit()
             # patch the api key and make it expired
             new_auth_token.expires = 0
             Session().add(new_auth_token)
             Session().commit()
             with fixture.anon_access(False):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  repo_name=HG_REPO, revision='tip',
                                  api_key=new_auth_token.api_key),
                              status=302)

rhodecode/tests/functional/test_pullrequests.py

@@ -1,916 +1,917 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import mock
 import pytest
 from webob.exc import HTTPNotFound
 
 import rhodecode
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.model.changeset_status import ChangesetStatusModel
 from rhodecode.model.db import (
     PullRequest, ChangesetStatus, UserLog, Notification)
 from rhodecode.model.meta import Session
 from rhodecode.model.pull_request import PullRequestModel
 from rhodecode.model.user import UserModel
 from rhodecode.tests import assert_session_flash, url, TEST_USER_ADMIN_LOGIN
 from rhodecode.tests.utils import AssertResponse
 
 
 @pytest.mark.usefixtures('app', 'autologin_user')
 @pytest.mark.backends("git", "hg")
 class TestPullrequestsController:
 
     def test_index(self, backend):
         self.app.get(url(
             controller='pullrequests', action='index',
             repo_name=backend.repo_name))
 
     def test_option_menu_create_pull_request_exists(self, backend):
         repo_name = backend.repo_name
         response = self.app.get(url('summary_home', repo_name=repo_name))
 
         create_pr_link = '<a href="%s">Create Pull Request</a>' % url(
             'pullrequest', repo_name=repo_name)
         response.mustcontain(create_pr_link)
 
     def test_global_redirect_of_pr(self, backend, pr_util):
         pull_request = pr_util.create_pull_request()
 
         response = self.app.get(
             url('pull_requests_global',
                 pull_request_id=pull_request.pull_request_id))
 
         repo_name = pull_request.target_repo.repo_name
         redirect_url = url('pullrequest_show', repo_name=repo_name,
                            pull_request_id=pull_request.pull_request_id)
         assert response.status == '302 Found'
         assert redirect_url in response.location
 
     @pytest.mark.xfail_backends(
         "git", reason="Pending bugfix/feature, issue #6")
     def test_create_pr_form_with_raw_commit_id(self, backend):
         repo = backend.repo
 
         self.app.get(
             url(controller='pullrequests', action='index',
                 repo_name=repo.repo_name,
                 commit=repo.get_commit().raw_id),
             status=200)
 
     @pytest.mark.parametrize('pr_merge_enabled', [True, False])
     def test_show(self, pr_util, pr_merge_enabled):
         pull_request = pr_util.create_pull_request(
             mergeable=pr_merge_enabled, enable_notifications=False)
 
         response = self.app.get(url(
             controller='pullrequests', action='show',
             repo_name=pull_request.target_repo.scm_instance().name,
             pull_request_id=str(pull_request.pull_request_id)))
 
         for commit_id in pull_request.revisions:
             response.mustcontain(commit_id)
 
         assert pull_request.target_ref_parts.type in response
         assert pull_request.target_ref_parts.name in response
         target_clone_url = pull_request.target_repo.clone_url()
         assert target_clone_url in response
 
         assert 'class="pull-request-merge"' in response
         assert (
             'Server-side pull request merging is disabled.'
             in response) != pr_merge_enabled
 
     def test_close_status_visibility(self, pr_util, csrf_token):
+        from rhodecode.tests.functional.test_login import login_url, logut_url
         # Logout
         response = self.app.post(
-            url(controller='login', action='logout'),
+            logut_url,
             params={'csrf_token': csrf_token})
         # Login as regular user
-        response = self.app.post(url(controller='login', action='index'),
+        response = self.app.post(login_url,
                                  {'username': 'test_regular',
                                   'password': 'test12'})
 
         pull_request = pr_util.create_pull_request(author='test_regular')
 
         response = self.app.get(url(
             controller='pullrequests', action='show',
             repo_name=pull_request.target_repo.scm_instance().name,
             pull_request_id=str(pull_request.pull_request_id)))
 
         assert 'Server-side pull request merging is disabled.' in response
         assert 'value="forced_closed"' in response
 
     def test_show_invalid_commit_id(self, pr_util):
         # Simulating invalid revisions which will cause a lookup error
         pull_request = pr_util.create_pull_request()
         pull_request.revisions = ['invalid']
         Session().add(pull_request)
         Session().commit()
 
         response = self.app.get(url(
             controller='pullrequests', action='show',
             repo_name=pull_request.target_repo.scm_instance().name,
             pull_request_id=str(pull_request.pull_request_id)))
 
         for commit_id in pull_request.revisions:
             response.mustcontain(commit_id)
 
     def test_show_invalid_source_reference(self, pr_util):
         pull_request = pr_util.create_pull_request()
         pull_request.source_ref = 'branch:b:invalid'
         Session().add(pull_request)
         Session().commit()
 
         self.app.get(url(
             controller='pullrequests', action='show',
             repo_name=pull_request.target_repo.scm_instance().name,
             pull_request_id=str(pull_request.pull_request_id)))
 
     def test_edit_title_description(self, pr_util, csrf_token):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
 
         response = self.app.post(
             url(controller='pullrequests', action='update',
                 repo_name=pull_request.target_repo.repo_name,
                 pull_request_id=str(pull_request_id)),
             params={
                 'edit_pull_request': 'true',
                 '_method': 'put',
                 'title': 'New title',
                 'description': 'New description',
                 'csrf_token': csrf_token})
 
         assert_session_flash(
             response, u'Pull request title & description updated.',
             category='success')
 
         pull_request = PullRequest.get(pull_request_id)
         assert pull_request.title == 'New title'
         assert pull_request.description == 'New description'
 
     def test_edit_title_description_closed(self, pr_util, csrf_token):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
         pr_util.close()
 
         response = self.app.post(
             url(controller='pullrequests', action='update',
                 repo_name=pull_request.target_repo.repo_name,
                 pull_request_id=str(pull_request_id)),
             params={
                 'edit_pull_request': 'true',
                 '_method': 'put',
                 'title': 'New title',
                 'description': 'New description',
                 'csrf_token': csrf_token})
 
         assert_session_flash(
             response, u'Cannot update closed pull requests.',
             category='error')
 
     def test_update_invalid_source_reference(self, pr_util, csrf_token):
         pull_request = pr_util.create_pull_request()
         pull_request.source_ref = 'branch:invalid-branch:invalid-commit-id'
         Session().add(pull_request)
         Session().commit()
 
         pull_request_id = pull_request.pull_request_id
 
         response = self.app.post(
             url(controller='pullrequests', action='update',
                 repo_name=pull_request.target_repo.repo_name,
                 pull_request_id=str(pull_request_id)),
             params={'update_commits': 'true', '_method': 'put',
                     'csrf_token': csrf_token})
 
         assert_session_flash(
             response, u'Update failed due to missing commits.',
             category='error')
 
     def test_comment_and_close_pull_request(self, pr_util, csrf_token):
         pull_request = pr_util.create_pull_request(approved=True)
         pull_request_id = pull_request.pull_request_id
         author = pull_request.user_id
         repo = pull_request.target_repo.repo_id
 
         self.app.post(
             url(controller='pullrequests',
                 action='comment',
                 repo_name=pull_request.target_repo.scm_instance().name,
                 pull_request_id=str(pull_request_id)),
             params={
                 'changeset_status':
                     ChangesetStatus.STATUS_APPROVED + '_closed',
                 'change_changeset_status': 'on',
                 'text': '',
                 'csrf_token': csrf_token},
             status=302)
 
         action = 'user_closed_pull_request:%d' % pull_request_id
         journal = UserLog.query()\
             .filter(UserLog.user_id == author)\
             .filter(UserLog.repository_id == repo)\
             .filter(UserLog.action == action)\
             .all()
         assert len(journal) == 1
 
     def test_reject_and_close_pull_request(self, pr_util, csrf_token):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
         response = self.app.post(
             url(controller='pullrequests',
                 action='update',
                 repo_name=pull_request.target_repo.scm_instance().name,
                 pull_request_id=str(pull_request.pull_request_id)),
             params={'close_pull_request': 'true', '_method': 'put',
                     'csrf_token': csrf_token})
 
         pull_request = PullRequest.get(pull_request_id)
 
         assert response.json is True
         assert pull_request.is_closed()
 
         # check only the latest status, not the review status
         status = ChangesetStatusModel().get_status(
             pull_request.source_repo, pull_request=pull_request)
         assert status == ChangesetStatus.STATUS_REJECTED
 
     def test_comment_force_close_pull_request(self, pr_util, csrf_token):
         pull_request = pr_util.create_pull_request()
         pull_request_id = pull_request.pull_request_id
         reviewers_ids = [1, 2]
         PullRequestModel().update_reviewers(pull_request_id, reviewers_ids)
         author = pull_request.user_id
         repo = pull_request.target_repo.repo_id
         self.app.post(
             url(controller='pullrequests',
                 action='comment',
                 repo_name=pull_request.target_repo.scm_instance().name,
                 pull_request_id=str(pull_request_id)),
             params={
                 'changeset_status': 'forced_closed',
                 'csrf_token': csrf_token},
             status=302)
 
         pull_request = PullRequest.get(pull_request_id)
 
         action = 'user_closed_pull_request:%d' % pull_request_id
         journal = UserLog.query().filter(
             UserLog.user_id == author,
             UserLog.repository_id == repo,
             UserLog.action == action).all()
         assert len(journal) == 1
 
         # check only the latest status, not the review status
         status = ChangesetStatusModel().get_status(
             pull_request.source_repo, pull_request=pull_request)
         assert status == ChangesetStatus.STATUS_REJECTED
 
     def test_create_pull_request(self, backend, csrf_token):
         commits = [
             {'message': 'ancestor'},
             {'message': 'change'},
         ]
         commit_ids = backend.create_master_repo(commits)
         target = backend.create_repo(heads=['ancestor'])
         source = backend.create_repo(heads=['change'])
 
         response = self.app.post(
             url(
                 controller='pullrequests',
                 action='create',
                 repo_name=source.repo_name),
             params={
                 'source_repo': source.repo_name,
                 'source_ref': 'branch:default:' + commit_ids['change'],
                 'target_repo': target.repo_name,
                 'target_ref': 'branch:default:' + commit_ids['ancestor'],
                 'pullrequest_desc': 'Description',
                 'pullrequest_title': 'Title',
                 'review_members': '1',
                 'revisions': commit_ids['change'],
                 'user': '',
                 'csrf_token': csrf_token,
             },
             status=302)
 
         location = response.headers['Location']
         pull_request_id = int(location.rsplit('/', 1)[1])
         pull_request = PullRequest.get(pull_request_id)
 
         # check that we have now both revisions
         assert pull_request.revisions == [commit_ids['change']]
         assert pull_request.source_ref == 'branch:default:' + commit_ids['change']
         expected_target_ref = 'branch:default:' + commit_ids['ancestor']
         assert pull_request.target_ref == expected_target_ref
 
     def test_reviewer_notifications(self, backend, csrf_token):
         # We have to use the app.post for this test so it will create the
         # notifications properly with the new PR
         commits = [
             {'message': 'ancestor',
              'added': [FileNode('file_A', content='content_of_ancestor')]},
             {'message': 'change',
              'added': [FileNode('file_a', content='content_of_change')]},
             {'message': 'change-child'},
             {'message': 'ancestor-child', 'parents': ['ancestor'],
              'added': [
                 FileNode('file_B', content='content_of_ancestor_child')]},
             {'message': 'ancestor-child-2'},
         ]
         commit_ids = backend.create_master_repo(commits)
         target = backend.create_repo(heads=['ancestor-child'])
         source = backend.create_repo(heads=['change'])
 
         response = self.app.post(
             url(
                 controller='pullrequests',
                 action='create',
                 repo_name=source.repo_name),
             params={
                 'source_repo': source.repo_name,
                 'source_ref': 'branch:default:' + commit_ids['change'],
                 'target_repo': target.repo_name,
                 'target_ref': 'branch:default:' + commit_ids['ancestor-child'],
                 'pullrequest_desc': 'Description',
                 'pullrequest_title': 'Title',
                 'review_members': '2',
                 'revisions': commit_ids['change'],
                 'user': '',
                 'csrf_token': csrf_token,
             },
             status=302)
 
         location = response.headers['Location']
         pull_request_id = int(location.rsplit('/', 1)[1])
         pull_request = PullRequest.get(pull_request_id)
 
         # Check that a notification was made
         notifications = Notification.query()\
             .filter(Notification.created_by == pull_request.author.user_id,
                     Notification.type_ == Notification.TYPE_PULL_REQUEST,
                     Notification.subject.contains("wants you to review "
                                                   "pull request #%d"
                                                   % pull_request_id))
         assert len(notifications.all()) == 1
 
         # Change reviewers and check that a notification was made
         PullRequestModel().update_reviewers(pull_request.pull_request_id, [1])
         assert len(notifications.all()) == 2
 
     def test_create_pull_request_stores_ancestor_commit_id(self, backend,
                                                            csrf_token):
         commits = [
             {'message': 'ancestor',
              'added': [FileNode('file_A', content='content_of_ancestor')]},
             {'message': 'change',
              'added': [FileNode('file_a', content='content_of_change')]},
             {'message': 'change-child'},
             {'message': 'ancestor-child', 'parents': ['ancestor'],
              'added': [
                 FileNode('file_B', content='content_of_ancestor_child')]},
             {'message': 'ancestor-child-2'},
         ]
         commit_ids = backend.create_master_repo(commits)
         target = backend.create_repo(heads=['ancestor-child'])
         source = backend.create_repo(heads=['change'])
 
         response = self.app.post(
             url(
                 controller='pullrequests',
                 action='create',
                 repo_name=source.repo_name),
             params={
                 'source_repo': source.repo_name,
                 'source_ref': 'branch:default:' + commit_ids['change'],
                 'target_repo': target.repo_name,
                 'target_ref': 'branch:default:' + commit_ids['ancestor-child'],
                 'pullrequest_desc': 'Description',
                 'pullrequest_title': 'Title',
                 'review_members': '1',
                 'revisions': commit_ids['change'],
                 'user': '',
                 'csrf_token': csrf_token,
             },
             status=302)
 
         location = response.headers['Location']
         pull_request_id = int(location.rsplit('/', 1)[1])
         pull_request = PullRequest.get(pull_request_id)
 
         # target_ref has to point to the ancestor's commit_id in order to
         # show the correct diff
         expected_target_ref = 'branch:default:' + commit_ids['ancestor']
         assert pull_request.target_ref == expected_target_ref
 
         # Check generated diff contents
         response = response.follow()
         assert 'content_of_ancestor' not in response.body
         assert 'content_of_ancestor-child' not in response.body
         assert 'content_of_change' in response.body
 
     def test_merge_pull_request_enabled(self, pr_util, csrf_token):
         # Clear any previous calls to rcextensions
         rhodecode.EXTENSIONS.calls.clear()
 
         pull_request = pr_util.create_pull_request(
             approved=True, mergeable=True)
         pull_request_id = pull_request.pull_request_id
         repo_name = pull_request.target_repo.scm_instance().name,
 
         response = self.app.post(
             url(controller='pullrequests',
                 action='merge',
                 repo_name=str(repo_name[0]),
                 pull_request_id=str(pull_request_id)),
             params={'csrf_token': csrf_token}).follow()
 
         pull_request = PullRequest.get(pull_request_id)
 
         assert response.status_int == 200
         assert pull_request.is_closed()
         assert_pull_request_status(
             pull_request, ChangesetStatus.STATUS_APPROVED)
 
         # Check the relevant log entries were added
         user_logs = UserLog.query().order_by('-user_log_id').limit(4)
         actions = [log.action for log in user_logs]
         pr_commit_ids = PullRequestModel()._get_commit_ids(pull_request)
         expected_actions = [
             u'user_closed_pull_request:%d' % pull_request_id,
             u'user_merged_pull_request:%d' % pull_request_id,
             # The action below reflect that the post push actions were executed
             u'user_commented_pull_request:%d' % pull_request_id,
             u'push:%s' % ','.join(pr_commit_ids),
         ]
         assert actions == expected_actions
 
         # Check post_push rcextension was really executed
         push_calls = rhodecode.EXTENSIONS.calls['post_push']
         assert len(push_calls) == 1
         unused_last_call_args, last_call_kwargs = push_calls[0]
         assert last_call_kwargs['action'] == 'push'
         assert last_call_kwargs['pushed_revs'] == pr_commit_ids
 
     def test_merge_pull_request_disabled(self, pr_util, csrf_token):
         pull_request = pr_util.create_pull_request(mergeable=False)
         pull_request_id = pull_request.pull_request_id
         pull_request = PullRequest.get(pull_request_id)
 
         response = self.app.post(
             url(controller='pullrequests',
                 action='merge',
                 repo_name=pull_request.target_repo.scm_instance().name,
                 pull_request_id=str(pull_request.pull_request_id)),
             params={'csrf_token': csrf_token}).follow()
 
         assert response.status_int == 200
         assert 'Server-side pull request merging is disabled.' in response.body
 
     @pytest.mark.skip_backends('svn')
     def test_merge_pull_request_not_approved(self, pr_util, csrf_token):
         pull_request = pr_util.create_pull_request(mergeable=True)
         pull_request_id = pull_request.pull_request_id
         repo_name = pull_request.target_repo.scm_instance().name,
 
         response = self.app.post(
             url(controller='pullrequests',
                 action='merge',
                 repo_name=str(repo_name[0]),
                 pull_request_id=str(pull_request_id)),
             params={'csrf_token': csrf_token}).follow()
 
         pull_request = PullRequest.get(pull_request_id)
 
         assert response.status_int == 200
         assert ' Reviewer approval is pending.' in response.body
 
     def test_update_source_revision(self, backend, csrf_token):
         commits = [
             {'message': 'ancestor'},
             {'message': 'change'},
             {'message': 'change-2'},
         ]
         commit_ids = backend.create_master_repo(commits)
         target = backend.create_repo(heads=['ancestor'])
         source = backend.create_repo(heads=['change'])
 
         # create pr from a in source to A in target
         pull_request = PullRequest()
         pull_request.source_repo = source
         # TODO: johbo: Make sure that we write the source ref this way!
         pull_request.source_ref = 'branch:{branch}:{commit_id}'.format(
             branch=backend.default_branch_name, commit_id=commit_ids['change'])
         pull_request.target_repo = target
 
         pull_request.target_ref = 'branch:{branch}:{commit_id}'.format(
             branch=backend.default_branch_name,
             commit_id=commit_ids['ancestor'])
         pull_request.revisions = [commit_ids['change']]
         pull_request.title = u"Test"
         pull_request.description = u"Description"
         pull_request.author = UserModel().get_by_username(
             TEST_USER_ADMIN_LOGIN)
         Session().add(pull_request)
         Session().commit()
         pull_request_id = pull_request.pull_request_id
 
         # source has ancestor - change - change-2
         backend.pull_heads(source, heads=['change-2'])
 
         # update PR
         self.app.post(
             url(controller='pullrequests', action='update',
                 repo_name=target.repo_name,
                 pull_request_id=str(pull_request_id)),
             params={'update_commits': 'true', '_method': 'put',
                     'csrf_token': csrf_token})
 
         # check that we have now both revisions
         pull_request = PullRequest.get(pull_request_id)
         assert pull_request.revisions == [
             commit_ids['change-2'], commit_ids['change']]
 
         # TODO: johbo: this should be a test on its own
         response = self.app.get(url(
             controller='pullrequests', action='index',
             repo_name=target.repo_name))
         assert response.status_int == 200
         assert 'Pull request updated to' in response.body
         assert 'with 1 added, 0 removed commits.' in response.body
 
     def test_update_target_revision(self, backend, csrf_token):
         commits = [
             {'message': 'ancestor'},
             {'message': 'change'},
             {'message': 'ancestor-new', 'parents': ['ancestor']},
             {'message': 'change-rebased'},
         ]
         commit_ids = backend.create_master_repo(commits)
         target = backend.create_repo(heads=['ancestor'])
         source = backend.create_repo(heads=['change'])
 
         # create pr from a in source to A in target
         pull_request = PullRequest()
         pull_request.source_repo = source
         # TODO: johbo: Make sure that we write the source ref this way!
         pull_request.source_ref = 'branch:{branch}:{commit_id}'.format(
             branch=backend.default_branch_name, commit_id=commit_ids['change'])
         pull_request.target_repo = target
         # TODO: johbo: Target ref should be branch based, since tip can jump
         # from branch to branch
         pull_request.target_ref = 'branch:{branch}:{commit_id}'.format(
             branch=backend.default_branch_name,
             commit_id=commit_ids['ancestor'])
         pull_request.revisions = [commit_ids['change']]
         pull_request.title = u"Test"
         pull_request.description = u"Description"
         pull_request.author = UserModel().get_by_username(
             TEST_USER_ADMIN_LOGIN)
         Session().add(pull_request)
         Session().commit()
         pull_request_id = pull_request.pull_request_id
 
         # target has ancestor - ancestor-new
         # source has ancestor - ancestor-new - change-rebased
         backend.pull_heads(target, heads=['ancestor-new'])
         backend.pull_heads(source, heads=['change-rebased'])
 
         # update PR
         self.app.post(
             url(controller='pullrequests', action='update',
                 repo_name=target.repo_name,
                 pull_request_id=str(pull_request_id)),
             params={'update_commits': 'true', '_method': 'put',
                     'csrf_token': csrf_token},
             status=200)
 
         # check that we have now both revisions
         pull_request = PullRequest.get(pull_request_id)
         assert pull_request.revisions == [commit_ids['change-rebased']]
         assert pull_request.target_ref == 'branch:{branch}:{commit_id}'.format(
             branch=backend.default_branch_name,
             commit_id=commit_ids['ancestor-new'])
 
         # TODO: johbo: This should be a test on its own
         response = self.app.get(url(
             controller='pullrequests', action='index',
             repo_name=target.repo_name))
         assert response.status_int == 200
         assert 'Pull request updated to' in response.body
         assert 'with 1 added, 1 removed commits.' in response.body
 
     def test_update_of_ancestor_reference(self, backend, csrf_token):
         commits = [
             {'message': 'ancestor'},
             {'message': 'change'},
             {'message': 'change-2'},
             {'message': 'ancestor-new', 'parents': ['ancestor']},
             {'message': 'change-rebased'},
         ]
         commit_ids = backend.create_master_repo(commits)
         target = backend.create_repo(heads=['ancestor'])
         source = backend.create_repo(heads=['change'])
 
         # create pr from a in source to A in target
         pull_request = PullRequest()
         pull_request.source_repo = source
         # TODO: johbo: Make sure that we write the source ref this way!
         pull_request.source_ref = 'branch:{branch}:{commit_id}'.format(
             branch=backend.default_branch_name,
             commit_id=commit_ids['change'])
         pull_request.target_repo = target
         # TODO: johbo: Target ref should be branch based, since tip can jump
         # from branch to branch
         pull_request.target_ref = 'branch:{branch}:{commit_id}'.format(
             branch=backend.default_branch_name,
             commit_id=commit_ids['ancestor'])
         pull_request.revisions = [commit_ids['change']]
         pull_request.title = u"Test"
         pull_request.description = u"Description"
         pull_request.author = UserModel().get_by_username(
             TEST_USER_ADMIN_LOGIN)
         Session().add(pull_request)
         Session().commit()
         pull_request_id = pull_request.pull_request_id
 
         # target has ancestor - ancestor-new
         # source has ancestor - ancestor-new - change-rebased
         backend.pull_heads(target, heads=['ancestor-new'])
         backend.pull_heads(source, heads=['change-rebased'])
 
         # update PR
         self.app.post(
             url(controller='pullrequests', action='update',
                 repo_name=target.repo_name,
                 pull_request_id=str(pull_request_id)),
             params={'update_commits': 'true', '_method': 'put',
                     'csrf_token': csrf_token},
             status=200)
 
         # Expect the target reference to be updated correctly
         pull_request = PullRequest.get(pull_request_id)
         assert pull_request.revisions == [commit_ids['change-rebased']]
         expected_target_ref = 'branch:{branch}:{commit_id}'.format(
             branch=backend.default_branch_name,
             commit_id=commit_ids['ancestor-new'])
         assert pull_request.target_ref == expected_target_ref
 
     def test_remove_pull_request_branch(self, backend_git, csrf_token):
         branch_name = 'development'
         commits = [
             {'message': 'initial-commit'},
             {'message': 'old-feature'},
             {'message': 'new-feature', 'branch': branch_name},
         ]
         repo = backend_git.create_repo(commits)
         commit_ids = backend_git.commit_ids
 
         pull_request = PullRequest()
         pull_request.source_repo = repo
         pull_request.target_repo = repo
         pull_request.source_ref = 'branch:{branch}:{commit_id}'.format(
             branch=branch_name, commit_id=commit_ids['new-feature'])
         pull_request.target_ref = 'branch:{branch}:{commit_id}'.format(
             branch=backend_git.default_branch_name,
             commit_id=commit_ids['old-feature'])
         pull_request.revisions = [commit_ids['new-feature']]
         pull_request.title = u"Test"
         pull_request.description = u"Description"
         pull_request.author = UserModel().get_by_username(
             TEST_USER_ADMIN_LOGIN)
         Session().add(pull_request)
         Session().commit()
 
         vcs = repo.scm_instance()
         vcs.remove_ref('refs/heads/{}'.format(branch_name))
 
         response = self.app.get(url(
             controller='pullrequests', action='show',
             repo_name=repo.repo_name,
             pull_request_id=str(pull_request.pull_request_id)))
 
         assert response.status_int == 200
         assert_response = AssertResponse(response)
         assert_response.element_contains(
             '#changeset_compare_view_content .alert strong',
             'Missing commits')
         assert_response.element_contains(
             '#changeset_compare_view_content .alert',
             'This pull request cannot be displayed, because one or more'
             ' commits no longer exist in the source repository.')
 
     def test_strip_commits_from_pull_request(
             self, backend, pr_util, csrf_token):
         commits = [
             {'message': 'initial-commit'},
             {'message': 'old-feature'},
             {'message': 'new-feature', 'parents': ['initial-commit']},
         ]
         pull_request = pr_util.create_pull_request(
             commits, target_head='initial-commit', source_head='new-feature',
             revisions=['new-feature'])
 
         vcs = pr_util.source_repository.scm_instance()
         if backend.alias == 'git':
             vcs.strip(pr_util.commit_ids['new-feature'], branch_name='master')
         else:
             vcs.strip(pr_util.commit_ids['new-feature'])
 
         response = self.app.get(url(
             controller='pullrequests', action='show',
             repo_name=pr_util.target_repository.repo_name,
             pull_request_id=str(pull_request.pull_request_id)))
 
         assert response.status_int == 200
         assert_response = AssertResponse(response)
         assert_response.element_contains(
             '#changeset_compare_view_content .alert strong',
             'Missing commits')
         assert_response.element_contains(
             '#changeset_compare_view_content .alert',
             'This pull request cannot be displayed, because one or more'
             ' commits no longer exist in the source repository.')
         assert_response.element_contains(
             '#update_commits',
             'Update commits')
 
     def test_strip_commits_and_update(
             self, backend, pr_util, csrf_token):
         commits = [
             {'message': 'initial-commit'},
             {'message': 'old-feature'},
             {'message': 'new-feature', 'parents': ['old-feature']},
         ]
         pull_request = pr_util.create_pull_request(
             commits, target_head='old-feature', source_head='new-feature',
             revisions=['new-feature'], mergeable=True)
 
         vcs = pr_util.source_repository.scm_instance()
         if backend.alias == 'git':
             vcs.strip(pr_util.commit_ids['new-feature'], branch_name='master')
         else:
             vcs.strip(pr_util.commit_ids['new-feature'])
 
         response = self.app.post(
             url(controller='pullrequests', action='update',
                 repo_name=pull_request.target_repo.repo_name,
                 pull_request_id=str(pull_request.pull_request_id)),
             params={'update_commits': 'true', '_method': 'put',
                     'csrf_token': csrf_token})
 
         assert response.status_int == 200
         assert response.body == 'true'
 
         # Make sure that after update, it won't raise 500 errors
         response = self.app.get(url(
             controller='pullrequests', action='show',
             repo_name=pr_util.target_repository.repo_name,
             pull_request_id=str(pull_request.pull_request_id)))
 
         assert response.status_int == 200
         assert_response = AssertResponse(response)
         assert_response.element_contains(
             '#changeset_compare_view_content .alert strong',
             'Missing commits')
 
     def test_branch_is_a_link(self, pr_util):
         pull_request = pr_util.create_pull_request()
         pull_request.source_ref = 'branch:origin:1234567890abcdef'
         pull_request.target_ref = 'branch:target:abcdef1234567890'
         Session().add(pull_request)
         Session().commit()
 
         response = self.app.get(url(
             controller='pullrequests', action='show',
             repo_name=pull_request.target_repo.scm_instance().name,
             pull_request_id=str(pull_request.pull_request_id)))
         assert response.status_int == 200
         assert_response = AssertResponse(response)
 
         origin = assert_response.get_element('.pr-origininfo .tag')
         origin_children = origin.getchildren()
         assert len(origin_children) == 1
         target = assert_response.get_element('.pr-targetinfo .tag')
         target_children = target.getchildren()
         assert len(target_children) == 1
 
         expected_origin_link = url(
             'changelog_home',
             repo_name=pull_request.source_repo.scm_instance().name,
             branch='origin')
         expected_target_link = url(
             'changelog_home',
             repo_name=pull_request.target_repo.scm_instance().name,
             branch='target')
         assert origin_children[0].attrib['href'] == expected_origin_link
         assert origin_children[0].text == 'branch: origin'
         assert target_children[0].attrib['href'] == expected_target_link
         assert target_children[0].text == 'branch: target'
 
     def test_bookmark_is_not_a_link(self, pr_util):
         pull_request = pr_util.create_pull_request()
         pull_request.source_ref = 'bookmark:origin:1234567890abcdef'
         pull_request.target_ref = 'bookmark:target:abcdef1234567890'
         Session().add(pull_request)
         Session().commit()
 
         response = self.app.get(url(
             controller='pullrequests', action='show',
             repo_name=pull_request.target_repo.scm_instance().name,
             pull_request_id=str(pull_request.pull_request_id)))
         assert response.status_int == 200
         assert_response = AssertResponse(response)
 
         origin = assert_response.get_element('.pr-origininfo .tag')
         assert origin.text.strip() == 'bookmark: origin'
         assert origin.getchildren() == []
 
         target = assert_response.get_element('.pr-targetinfo .tag')
         assert target.text.strip() == 'bookmark: target'
         assert target.getchildren() == []
 
     def test_tag_is_not_a_link(self, pr_util):
         pull_request = pr_util.create_pull_request()
         pull_request.source_ref = 'tag:origin:1234567890abcdef'
         pull_request.target_ref = 'tag:target:abcdef1234567890'
         Session().add(pull_request)
         Session().commit()
 
         response = self.app.get(url(
             controller='pullrequests', action='show',
             repo_name=pull_request.target_repo.scm_instance().name,
             pull_request_id=str(pull_request.pull_request_id)))
         assert response.status_int == 200
         assert_response = AssertResponse(response)
 
         origin = assert_response.get_element('.pr-origininfo .tag')
         assert origin.text.strip() == 'tag: origin'
         assert origin.getchildren() == []
 
         target = assert_response.get_element('.pr-targetinfo .tag')
         assert target.text.strip() == 'tag: target'
         assert target.getchildren() == []
 
     def test_description_is_escaped_on_index_page(self, backend, pr_util):
         xss_description = "<script>alert('Hi!')</script>"
         pull_request = pr_util.create_pull_request(description=xss_description)
         response = self.app.get(url(
             controller='pullrequests', action='show_all',
             repo_name=pull_request.target_repo.repo_name))
         response.mustcontain(
             "&lt;script&gt;alert(&#39;Hi!&#39;)&lt;/script&gt;")
 
 
 def assert_pull_request_status(pull_request, expected_status):
     status = ChangesetStatusModel().calculated_review_status(
         pull_request=pull_request)
     assert status == expected_status
 
 
 @pytest.mark.parametrize('action', ['show_all', 'index', 'create'])
 @pytest.mark.usefixtures("autologin_user")
 def test_redirects_to_repo_summary_for_svn_repositories(
         backend_svn, app, action):
     denied_actions = ['show_all', 'index', 'create']
     for action in denied_actions:
         response = app.get(url(
             controller='pullrequests', action=action,
             repo_name=backend_svn.repo_name))
         assert response.status_int == 302
 
         # Not allowed, redirect to the summary
         redirected = response.follow()
         summary_url = url('summary_home', repo_name=backend_svn.repo_name)
 
         # URL adds leading slash and path doesn't have it
         assert redirected.req.path == summary_url
 
 
 def test_delete_comment_returns_404_if_comment_does_not_exist(pylonsapp):
     # TODO: johbo: Global import not possible because models.forms blows up
     from rhodecode.controllers.pullrequests import PullrequestsController
     controller = PullrequestsController()
     patcher = mock.patch(
         'rhodecode.model.db.BaseModel.get', return_value=None)
     with pytest.raises(HTTPNotFound), patcher:
         controller._delete_comment(1)