##// END OF EJS Templates
goto-switcher: optimized performance and query capabilities....
goto-switcher: optimized performance and query capabilities. - Previous implementation had on significant bug. The use of LIMIT 20 was limiting results BEFORE auth checks. In case of large ammount of similarly named repositories user didn't had access too, the result goto search was empty. This was becuase first 20 items fetched didn't pass permission checks and final auth list was empty. To fix this we now use proper filtering for auth using SQL. It means we first check user allowed repositories, and add this as a filter so end result from database is already to only the accessible repositories.

File last commit:

r1906:1eaf71e3 default
r2038:2bdf9d4d default
Show More
test_simplevcs.py
455 lines | 16.8 KiB | text/x-python | PythonLexer
project: added all source files and assets
r1 # -*- coding: utf-8 -*-
license: updated copyright year to 2017
r1271 # Copyright (C) 2010-2017 RhodeCode GmbH
project: added all source files and assets
r1 #
# 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 base64
import mock
import pytest
tests: use custom test app for better debug of .mustcontain function....
r1256
from rhodecode.tests.utils import CustomTestApp
project: added all source files and assets
r1
from rhodecode.lib.caching_query import FromCache
Martin Bornhold
tests: Adapt tests to recent changes.
r592 from rhodecode.lib.hooks_daemon import DummyHooksCallbackDaemon
project: added all source files and assets
r1 from rhodecode.lib.middleware import simplevcs
from rhodecode.lib.middleware.https_fixup import HttpsFixup
Martin Bornhold
tests: Fix test that expects the pyro4 scm app as default.
r975 from rhodecode.lib.middleware.utils import scm_app_http
project: added all source files and assets
r1 from rhodecode.model.db import User, _hash_key
from rhodecode.model.meta import Session
from rhodecode.tests import (
HG_REPO, TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS)
from rhodecode.tests.lib.middleware import mock_scm_app
class StubVCSController(simplevcs.SimpleVCS):
SCM = 'hg'
stub_response_body = tuple()
vcs: register repo_name as specialized middleware variables instead...
r757 def __init__(self, *args, **kwargs):
super(StubVCSController, self).__init__(*args, **kwargs)
Martin Bornhold
tests: Make return values of stub vcs controller editable.
r912 self._action = 'pull'
self._name = HG_REPO
self.set_repo_names(None)
vcs: register repo_name as specialized middleware variables instead...
r757
project: added all source files and assets
r1 def _get_repository_name(self, environ):
Martin Bornhold
tests: Make return values of stub vcs controller editable.
r912 return self._name
project: added all source files and assets
r1
def _get_action(self, environ):
Martin Bornhold
tests: Make return values of stub vcs controller editable.
r912 return self._action
project: added all source files and assets
r1
def _create_wsgi_app(self, repo_path, repo_name, config):
def fake_app(environ, start_response):
dependencies: bumped pyramid to 1.9 webob to 1.7.3 and webtest to 2.0.27...
r1906 headers = [
('Http-Accept', 'application/mercurial')
]
start_response('200 OK', headers)
project: added all source files and assets
r1 return self.stub_response_body
return fake_app
def _create_config(self, extras, repo_name):
return None
@pytest.fixture
def vcscontroller(pylonsapp, config_stub):
config_stub.testing_securitypolicy()
config_stub.include('rhodecode.authentication')
home: moved home and repo group views into pyramid....
r1774 #set_anonymous_access(True)
Martin Bornhold
tests: Adapt tests to recent changes.
r592 controller = StubVCSController(pylonsapp, pylonsapp.config, None)
project: added all source files and assets
r1 app = HttpsFixup(controller, pylonsapp.config)
tests: use custom test app for better debug of .mustcontain function....
r1256 app = CustomTestApp(app)
project: added all source files and assets
r1
_remove_default_user_from_query_cache()
# Sanity checks that things are set up correctly
app.get('/' + HG_REPO, status=200)
app.controller = controller
return app
def _remove_default_user_from_query_cache():
user = User.get_default_user(cache=True)
query = Session().query(User).filter(User.username == user.username)
caches: ensure we don't use non-ascii characters in cache keys....
r1749 query = query.options(
FromCache("sql_cache_short", "get_user_%s" % _hash_key(user.username)))
project: added all source files and assets
r1 query.invalidate()
Session().expire(user)
def test_handles_exceptions_during_permissions_checks(
vcscontroller, disable_anonymous_user):
user_and_pass = '%s:%s' % (TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS)
auth_password = base64.encodestring(user_and_pass).strip()
extra_environ = {
'AUTH_TYPE': 'Basic',
'HTTP_AUTHORIZATION': 'Basic %s' % auth_password,
'REMOTE_USER': TEST_USER_ADMIN_LOGIN,
}
# Verify that things are hooked up correctly
vcscontroller.get('/', status=200, extra_environ=extra_environ)
# Simulate trouble during permission checks
with mock.patch('rhodecode.model.db.User.get_by_username',
side_effect=Exception) as get_user:
# Verify that a correct 500 is returned and check that the expected
# code path was hit.
vcscontroller.get('/', status=500, extra_environ=extra_environ)
assert get_user.called
def test_returns_forbidden_if_no_anonymous_access(
vcscontroller, disable_anonymous_user):
vcscontroller.get('/', status=401)
class StubFailVCSController(simplevcs.SimpleVCS):
def _handle_request(self, environ, start_response):
raise Exception("BOOM")
@pytest.fixture(scope='module')
def fail_controller(pylonsapp):
Martin Bornhold
tests: Adapt tests to recent changes.
r592 controller = StubFailVCSController(pylonsapp, pylonsapp.config, None)
project: added all source files and assets
r1 controller = HttpsFixup(controller, pylonsapp.config)
tests: use custom test app for better debug of .mustcontain function....
r1256 controller = CustomTestApp(controller)
project: added all source files and assets
r1 return controller
def test_handles_exceptions_as_internal_server_error(fail_controller):
fail_controller.get('/', status=500)
def test_provides_traceback_for_appenlight(fail_controller):
response = fail_controller.get(
'/', status=500, extra_environ={'appenlight.client': 'fake'})
assert 'appenlight.__traceback' in response.request.environ
def test_provides_utils_scm_app_as_scm_app_by_default(pylonsapp):
Martin Bornhold
tests: Adapt tests to recent changes.
r592 controller = StubVCSController(pylonsapp, pylonsapp.config, None)
Martin Bornhold
tests: Fix test that expects the pyro4 scm app as default.
r975 assert controller.scm_app is scm_app_http
project: added all source files and assets
r1
def test_allows_to_override_scm_app_via_config(pylonsapp):
config = pylonsapp.config.copy()
config['vcs.scm_app_implementation'] = (
'rhodecode.tests.lib.middleware.mock_scm_app')
Martin Bornhold
tests: Adapt tests to recent changes.
r592 controller = StubVCSController(pylonsapp, config, None)
project: added all source files and assets
r1 assert controller.scm_app is mock_scm_app
@pytest.mark.parametrize('query_string, expected', [
('cmd=stub_command', True),
('cmd=listkeys', False),
])
def test_should_check_locking(query_string, expected):
result = simplevcs._should_check_locking(query_string)
assert result == expected
Martin Bornhold
tests: Add tests to check the shadow-repo regular expression.
r919 class TestShadowRepoRegularExpression(object):
Martin Bornhold
tests: Use only one place to define pr and shadow URL segments.
r920 pr_segment = 'pull-request'
shadow_segment = 'repository'
Martin Bornhold
tests: Add tests to check the shadow-repo regular expression.
r919 @pytest.mark.parametrize('url, expected', [
# repo with/without groups
Martin Bornhold
tests: Use only one place to define pr and shadow URL segments.
r920 ('My-Repo/{pr_segment}/1/{shadow_segment}', True),
('Group/My-Repo/{pr_segment}/2/{shadow_segment}', True),
('Group/Sub-Group/My-Repo/{pr_segment}/3/{shadow_segment}', True),
('Group/Sub-Group1/Sub-Group2/My-Repo/{pr_segment}/3/{shadow_segment}', True),
Martin Bornhold
tests: Add tests to check the shadow-repo regular expression.
r919
# pull request ID
Martin Bornhold
tests: Use only one place to define pr and shadow URL segments.
r920 ('MyRepo/{pr_segment}/1/{shadow_segment}', True),
('MyRepo/{pr_segment}/1234567890/{shadow_segment}', True),
('MyRepo/{pr_segment}/-1/{shadow_segment}', False),
('MyRepo/{pr_segment}/invalid/{shadow_segment}', False),
Martin Bornhold
tests: Add tests to check the shadow-repo regular expression.
r919
# unicode
Martin Bornhold
tests: Use only one place to define pr and shadow URL segments.
r920 (u'Sp€çîál-Repö/{pr_segment}/1/{shadow_segment}', True),
(u'Sp€çîál-Gröüp/Sp€çîál-Repö/{pr_segment}/1/{shadow_segment}', True),
Martin Bornhold
tests: Add tests to check the shadow-repo regular expression.
r919
# trailing/leading slash
Martin Bornhold
tests: Use only one place to define pr and shadow URL segments.
r920 ('/My-Repo/{pr_segment}/1/{shadow_segment}', False),
('My-Repo/{pr_segment}/1/{shadow_segment}/', False),
('/My-Repo/{pr_segment}/1/{shadow_segment}/', False),
Martin Bornhold
tests: Add tests to check the shadow-repo regular expression.
r919
# misc
Martin Bornhold
tests: Use only one place to define pr and shadow URL segments.
r920 ('My-Repo/{pr_segment}/1/{shadow_segment}/extra', False),
('My-Repo/{pr_segment}/1/{shadow_segment}extra', False),
Martin Bornhold
tests: Add tests to check the shadow-repo regular expression.
r919 ])
def test_shadow_repo_regular_expression(self, url, expected):
from rhodecode.lib.middleware.simplevcs import SimpleVCS
Martin Bornhold
tests: Use only one place to define pr and shadow URL segments.
r920 url = url.format(
pr_segment=self.pr_segment,
shadow_segment=self.shadow_segment)
Martin Bornhold
tests: Add tests to check the shadow-repo regular expression.
r919 match_obj = SimpleVCS.shadow_repo_re.match(url)
assert (match_obj is not None) == expected
Martin Bornhold
tests: Add tests to check the shadow repo exposure of the simplevcs class.
r913 @pytest.mark.backends('git', 'hg')
class TestShadowRepoExposure(object):
Martin Bornhold
tests: Add tests to check the shadow-repo regular expression.
r919
Martin Bornhold
tests: Add tests to check the shadow repo exposure of the simplevcs class.
r913 def test_pull_on_shadow_repo_propagates_to_wsgi_app(self, pylonsapp):
"""
Check that a pull action to a shadow repo is propagated to the
underlying wsgi app.
"""
controller = StubVCSController(pylonsapp, pylonsapp.config, None)
controller._check_ssl = mock.Mock()
controller.is_shadow_repo = True
controller._action = 'pull'
controller.stub_response_body = 'dummy body value'
environ_stub = {
'HTTP_HOST': 'test.example.com',
dependencies: bumped pyramid to 1.9 webob to 1.7.3 and webtest to 2.0.27...
r1906 'HTTP_ACCEPT': 'application/mercurial',
Martin Bornhold
tests: Add tests to check the shadow repo exposure of the simplevcs class.
r913 'REQUEST_METHOD': 'GET',
'wsgi.url_scheme': 'http',
}
response = controller(environ_stub, mock.Mock())
response_body = ''.join(response)
# Assert that we got the response from the wsgi app.
assert response_body == controller.stub_response_body
def test_push_on_shadow_repo_raises(self, pylonsapp):
"""
Check that a push action to a shadow repo is aborted.
"""
controller = StubVCSController(pylonsapp, pylonsapp.config, None)
controller._check_ssl = mock.Mock()
controller.is_shadow_repo = True
controller._action = 'push'
controller.stub_response_body = 'dummy body value'
environ_stub = {
'HTTP_HOST': 'test.example.com',
dependencies: bumped pyramid to 1.9 webob to 1.7.3 and webtest to 2.0.27...
r1906 'HTTP_ACCEPT': 'application/mercurial',
Martin Bornhold
tests: Add tests to check the shadow repo exposure of the simplevcs class.
r913 'REQUEST_METHOD': 'GET',
'wsgi.url_scheme': 'http',
}
response = controller(environ_stub, mock.Mock())
response_body = ''.join(response)
assert response_body != controller.stub_response_body
# Assert that a 406 error is returned.
assert '406 Not Acceptable' in response_body
def test_set_repo_names_no_shadow(self, pylonsapp):
"""
Check that the set_repo_names method sets all names to the one returned
by the _get_repository_name method on a request to a non shadow repo.
"""
environ_stub = {}
controller = StubVCSController(pylonsapp, pylonsapp.config, None)
controller._name = 'RepoGroup/MyRepo'
controller.set_repo_names(environ_stub)
assert not controller.is_shadow_repo
assert (controller.url_repo_name ==
controller.acl_repo_name ==
controller.vcs_repo_name ==
controller._get_repository_name(environ_stub))
home: moved home and repo group views into pyramid....
r1774 def test_set_repo_names_with_shadow(self, pylonsapp, pr_util, config_stub):
Martin Bornhold
tests: Add tests to check the shadow repo exposure of the simplevcs class.
r913 """
Check that the set_repo_names method sets correct names on a request
to a shadow repo.
"""
from rhodecode.model.pull_request import PullRequestModel
pull_request = pr_util.create_pull_request()
Martin Bornhold
tests: Use only one place to define pr and shadow URL segments.
r920 shadow_url = '{target}/{pr_segment}/{pr_id}/{shadow_segment}'.format(
Martin Bornhold
tests: Add tests to check the shadow repo exposure of the simplevcs class.
r913 target=pull_request.target_repo.repo_name,
Martin Bornhold
tests: Use only one place to define pr and shadow URL segments.
r920 pr_id=pull_request.pull_request_id,
pr_segment=TestShadowRepoRegularExpression.pr_segment,
shadow_segment=TestShadowRepoRegularExpression.shadow_segment)
Martin Bornhold
tests: Add tests to check the shadow repo exposure of the simplevcs class.
r913 controller = StubVCSController(pylonsapp, pylonsapp.config, None)
controller._name = shadow_url
controller.set_repo_names({})
# Get file system path to shadow repo for assertions.
workspace_id = PullRequestModel()._workspace_id(pull_request)
target_vcs = pull_request.target_repo.scm_instance()
vcs_repo_name = target_vcs._get_shadow_repository_path(
workspace_id)
assert controller.vcs_repo_name == vcs_repo_name
assert controller.url_repo_name == shadow_url
assert controller.acl_repo_name == pull_request.target_repo.repo_name
assert controller.is_shadow_repo
def test_set_repo_names_with_shadow_but_missing_pr(
home: moved home and repo group views into pyramid....
r1774 self, pylonsapp, pr_util, config_stub):
Martin Bornhold
tests: Add tests to check the shadow repo exposure of the simplevcs class.
r913 """
Checks that the set_repo_names method enforces matching target repos
and pull request IDs.
"""
pull_request = pr_util.create_pull_request()
Martin Bornhold
tests: Use only one place to define pr and shadow URL segments.
r920 shadow_url = '{target}/{pr_segment}/{pr_id}/{shadow_segment}'.format(
Martin Bornhold
tests: Add tests to check the shadow repo exposure of the simplevcs class.
r913 target=pull_request.target_repo.repo_name,
Martin Bornhold
tests: Use only one place to define pr and shadow URL segments.
r920 pr_id=999999999,
pr_segment=TestShadowRepoRegularExpression.pr_segment,
shadow_segment=TestShadowRepoRegularExpression.shadow_segment)
Martin Bornhold
tests: Add tests to check the shadow repo exposure of the simplevcs class.
r913 controller = StubVCSController(pylonsapp, pylonsapp.config, None)
controller._name = shadow_url
controller.set_repo_names({})
assert not controller.is_shadow_repo
assert (controller.url_repo_name ==
controller.acl_repo_name ==
controller.vcs_repo_name)
Martin Bornhold
tests: Fix tests which need the DB to run.
r915 @pytest.mark.usefixtures('db')
home: moved home and repo group views into pyramid....
r1774 class TestGenerateVcsResponse(object):
project: added all source files and assets
r1
def test_ensures_that_start_response_is_called_early_enough(self):
self.call_controller_with_response_body(iter(['a', 'b']))
assert self.start_response.called
def test_invalidates_cache_after_body_is_consumed(self):
result = self.call_controller_with_response_body(iter(['a', 'b']))
assert not self.was_cache_invalidated()
# Consume the result
list(result)
assert self.was_cache_invalidated()
@mock.patch('rhodecode.lib.middleware.simplevcs.HTTPLockedRC')
def test_handles_locking_exception(self, http_locked_rc):
result = self.call_controller_with_response_body(
self.raise_result_iter(vcs_kind='repo_locked'))
assert not http_locked_rc.called
# Consume the result
list(result)
assert http_locked_rc.called
@mock.patch('rhodecode.lib.middleware.simplevcs.HTTPRequirementError')
def test_handles_requirement_exception(self, http_requirement):
result = self.call_controller_with_response_body(
self.raise_result_iter(vcs_kind='requirement'))
assert not http_requirement.called
# Consume the result
list(result)
assert http_requirement.called
@mock.patch('rhodecode.lib.middleware.simplevcs.HTTPLockedRC')
def test_handles_locking_exception_in_app_call(self, http_locked_rc):
app_factory_patcher = mock.patch.object(
StubVCSController, '_create_wsgi_app')
with app_factory_patcher as app_factory:
app_factory().side_effect = self.vcs_exception()
result = self.call_controller_with_response_body(['a'])
list(result)
assert http_locked_rc.called
def test_raises_unknown_exceptions(self):
result = self.call_controller_with_response_body(
self.raise_result_iter(vcs_kind='unknown'))
with pytest.raises(Exception):
list(result)
def test_prepare_callback_daemon_is_called(self):
def side_effect(extras):
return DummyHooksCallbackDaemon(), extras
prepare_patcher = mock.patch.object(
StubVCSController, '_prepare_callback_daemon')
with prepare_patcher as prepare_mock:
prepare_mock.side_effect = side_effect
self.call_controller_with_response_body(iter(['a', 'b']))
assert prepare_mock.called
assert prepare_mock.call_count == 1
def call_controller_with_response_body(self, response_body):
Martin Bornhold
tests: Adapt tests to recent changes.
r592 settings = {
'base_path': 'fake_base_path',
'vcs.hooks.protocol': 'http',
'vcs.hooks.direct_calls': False,
}
controller = StubVCSController(None, settings, None)
project: added all source files and assets
r1 controller._invalidate_cache = mock.Mock()
controller.stub_response_body = response_body
self.start_response = mock.Mock()
result = controller._generate_vcs_response(
environ={}, start_response=self.start_response,
repo_path='fake_repo_path',
extras={}, action='push')
self.controller = controller
return result
def raise_result_iter(self, vcs_kind='repo_locked'):
"""
Simulates an exception due to a vcs raised exception if kind vcs_kind
"""
raise self.vcs_exception(vcs_kind=vcs_kind)
yield "never_reached"
def vcs_exception(self, vcs_kind='repo_locked'):
locked_exception = Exception('TEST_MESSAGE')
locked_exception._vcs_kind = vcs_kind
return locked_exception
def was_cache_invalidated(self):
return self.controller._invalidate_cache.called
home: moved home and repo group views into pyramid....
r1774 class TestInitializeGenerator(object):
project: added all source files and assets
r1
def test_drains_first_element(self):
gen = self.factory(['__init__', 1, 2])
result = list(gen)
assert result == [1, 2]
@pytest.mark.parametrize('values', [
[],
[1, 2],
])
def test_raises_value_error(self, values):
with pytest.raises(ValueError):
self.factory(values)
@simplevcs.initialize_generator
def factory(self, iterable):
for elem in iterable:
yield elem
class TestPrepareHooksDaemon(object):
Martin Bornhold
tests: Use settings dict from fixture instead of grabbing it from the WSGI app stack.
r603 def test_calls_imported_prepare_callback_daemon(self, app_settings):
project: added all source files and assets
r1 expected_extras = {'extra1': 'value1'}
daemon = DummyHooksCallbackDaemon()
Martin Bornhold
tests: Use settings dict from fixture instead of grabbing it from the WSGI app stack.
r603 controller = StubVCSController(None, app_settings, None)
project: added all source files and assets
r1 prepare_patcher = mock.patch.object(
simplevcs, 'prepare_callback_daemon',
return_value=(daemon, expected_extras))
with prepare_patcher as prepare_mock:
callback_daemon, extras = controller._prepare_callback_daemon(
expected_extras.copy())
prepare_mock.assert_called_once_with(
Martin Bornhold
tests: Adapt tests to recent changes.
r592 expected_extras,
Martin Bornhold
tests: Use settings dict from fixture instead of grabbing it from the WSGI app stack.
r603 protocol=app_settings['vcs.hooks.protocol'],
use_direct_calls=app_settings['vcs.hooks.direct_calls'])
project: added all source files and assets
r1
assert callback_daemon == daemon
assert extras == extras