tests: Adapt tests to recent changes.

rhodecode/tests/config/test_environment.py

@@ -1,115 +1,53 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 rhodecode.config import environment
 
 
-class TestUseDirectHookCalls(object):
-    @pytest.mark.parametrize('config', [
-        {
-            'vcs.hooks.direct_calls': 'true',
-            'base_path': 'fake_base_path'
-        }
-    ])
-    def test_returns_true_when_conditions_are_met(self, config):
-        result = environment._use_direct_hook_calls(config)
-        assert result is True
-
-    @pytest.mark.parametrize('config', [
-        {
-            'vcs.hooks.direct_calls': 'false',
-            'base_path': 'fake_base_path'
-        },
-        {
-            'base_path': 'fake_base_path'
-        }
-    ])
-    def test_returns_false_when_conditions_are_not_met(self, config):
-        result = environment._use_direct_hook_calls(config)
-        assert result is False
-
-
-class TestGetVcsHooksProtocol(object):
-    def test_returns_pyro4_by_default(self):
-        config = {}
-        result = environment._get_vcs_hooks_protocol(config)
-        assert result == 'pyro4'
-
-    @pytest.mark.parametrize('protocol', ['PYRO4', 'HTTP', 'Pyro4', 'Http'])
-    def test_returns_lower_case_value(self, protocol):
-        config = {
-            'vcs.hooks.protocol': protocol
-        }
-        result = environment._get_vcs_hooks_protocol(config)
-        assert result == protocol.lower()
-
-
-class TestLoadEnvironment(object):
-    def test_calls_use_direct_hook_calls(self, _external_calls_patcher):
-        global_conf = {
-            'here': '',
-            'vcs.connection_timeout': '0',
-            'vcs.server.enable': 'false'
-        }
-        app_conf = {
-            'cache_dir': '/tmp/',
-            '__file__': '/tmp/abcde.ini'
-        }
-        direct_calls_patcher = mock.patch.object(
-            environment, '_use_direct_hook_calls', return_value=True)
-        protocol_patcher = mock.patch.object(
-            environment, '_get_vcs_hooks_protocol', return_value='http')
-        with direct_calls_patcher as direct_calls_mock, \
-                protocol_patcher as protocol_mock:
-            environment.load_environment(global_conf, app_conf)
-        direct_calls_mock.call_count == 1
-        protocol_mock.call_count == 1
-
-
 @pytest.fixture
 def _external_calls_patcher(request):
     # TODO: mikhail: This is a temporary solution. Ideally load_environment
     # should be split into multiple small testable functions.
     utils_patcher = mock.patch.object(environment, 'utils')
 
     rhodecode_patcher = mock.patch.object(environment, 'rhodecode')
 
     db_config = mock.Mock()
     db_config.items.return_value = {
         'paths': [['/tmp/abc', '/tmp/def']]
     }
     db_config_patcher = mock.patch.object(
         environment, 'make_db_config', return_value=db_config)
 
     set_config_patcher = mock.patch.object(environment, 'set_rhodecode_config')
 
     utils_patcher.start()
     rhodecode_patcher.start()
     db_config_patcher.start()
     set_config_patcher.start()
 
     request.addfinalizer(utils_patcher.stop)
     request.addfinalizer(rhodecode_patcher.stop)
     request.addfinalizer(db_config_patcher.stop)
     request.addfinalizer(set_config_patcher.stop)

rhodecode/tests/lib/middleware/test_simplegit.py

@@ -1,107 +1,110 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 pytest
 import urlparse
 
 from rhodecode.tests.lib.middleware import mock_scm_app
 import rhodecode.lib.middleware.simplegit as simplegit
 
 
 def get_environ(url):
     """Construct a minimum WSGI environ based on the URL."""
     parsed_url = urlparse.urlparse(url)
     environ = {
         'PATH_INFO': parsed_url.path,
         'QUERY_STRING': parsed_url.query,
     }
 
     return environ
 
 
 @pytest.mark.parametrize(
     'url, expected_action',
     [
         ('/foo/bar/info/refs?service=git-upload-pack', 'pull'),
         ('/foo/bar/info/refs?service=git-receive-pack', 'push'),
         ('/foo/bar/git-upload-pack', 'pull'),
         ('/foo/bar/git-receive-pack', 'push'),
         # Edge case: missing data for info/refs
         ('/foo/info/refs?service=', 'pull'),
         ('/foo/info/refs', 'pull'),
         # Edge case: git command comes with service argument
         ('/foo/git-upload-pack?service=git-receive-pack', 'pull'),
         ('/foo/git-receive-pack?service=git-upload-pack', 'push'),
         # Edge case: repo name conflicts with git commands
         ('/git-receive-pack/git-upload-pack', 'pull'),
         ('/git-receive-pack/git-receive-pack', 'push'),
         ('/git-upload-pack/git-upload-pack', 'pull'),
         ('/git-upload-pack/git-receive-pack', 'push'),
         ('/foo/git-receive-pack', 'push'),
         # Edge case: not a smart protocol url
         ('/foo/bar', 'pull'),
     ])
-def test_get_action(url, expected_action):
+def test_get_action(url, expected_action, pylonsapp):
     app = simplegit.SimpleGit(application=None,
-                              config={'auth_ret_code': '', 'base_path': ''})
+                              config={'auth_ret_code': '', 'base_path': ''},
+                              registry=None)
     assert expected_action == app._get_action(get_environ(url))
 
 
 @pytest.mark.parametrize(
     'url, expected_repo_name',
     [
         ('/foo/info/refs?service=git-upload-pack', 'foo'),
         ('/foo/bar/info/refs?service=git-receive-pack', 'foo/bar'),
         ('/foo/git-upload-pack', 'foo'),
         ('/foo/git-receive-pack', 'foo'),
         ('/foo/bar/git-upload-pack', 'foo/bar'),
         ('/foo/bar/git-receive-pack', 'foo/bar'),
     ])
-def test_get_repository_name(url, expected_repo_name):
+def test_get_repository_name(url, expected_repo_name, pylonsapp):
     app = simplegit.SimpleGit(application=None,
-                              config={'auth_ret_code': '', 'base_path': ''})
+                              config={'auth_ret_code': '', 'base_path': ''},
+                              registry=None)
     assert expected_repo_name == app._get_repository_name(get_environ(url))
 
 
-def test_get_config():
+def test_get_config(pylonsapp):
     app = simplegit.SimpleGit(application=None,
-                              config={'auth_ret_code': '', 'base_path': ''})
+                              config={'auth_ret_code': '', 'base_path': ''},
+                              registry=None)
     extras = {'foo': 'FOO', 'bar': 'BAR'}
 
     # We copy the extras as the method below will change the contents.
     config = app._create_config(dict(extras), repo_name='test-repo')
     expected_config = dict(extras)
     expected_config.update({
         'git_update_server_info': False,
     })
 
     assert config == expected_config
 
 
-def test_create_wsgi_app_uses_scm_app_from_simplevcs():
+def test_create_wsgi_app_uses_scm_app_from_simplevcs(pylonsapp):
     config = {
         'auth_ret_code': '',
         'base_path': '',
         'vcs.scm_app_implementation':
             'rhodecode.tests.lib.middleware.mock_scm_app',
     }
-    app = simplegit.SimpleGit(application=None, config=config)
+    app = simplegit.SimpleGit(application=None, config=config, registry=None)
     wsgi_app = app._create_wsgi_app('/tmp/test', 'test_repo', {})
     assert wsgi_app is mock_scm_app.mock_git_wsgi

rhodecode/tests/lib/middleware/test_simplehg.py

@@ -1,113 +1,116 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
 import simplejson as json
 
 from rhodecode.lib.vcs.backends.base import Config
 from rhodecode.tests.lib.middleware import mock_scm_app
 import rhodecode.lib.middleware.simplehg as simplehg
 
 
 def get_environ(url):
     """Construct a minimum WSGI environ based on the URL."""
     parsed_url = urlparse.urlparse(url)
     environ = {
         'PATH_INFO': parsed_url.path,
         'QUERY_STRING': parsed_url.query,
     }
 
     return environ
 
 
 @pytest.mark.parametrize(
     'url, expected_action',
     [
         ('/foo/bar?cmd=unbundle&key=tip', 'push'),
         ('/foo/bar?cmd=pushkey&key=tip', 'push'),
         ('/foo/bar?cmd=listkeys&key=tip', 'pull'),
         ('/foo/bar?cmd=changegroup&key=tip', 'pull'),
         # Edge case: unknown argument: assume pull
         ('/foo/bar?cmd=unknown&key=tip', 'pull'),
         ('/foo/bar?cmd=&key=tip', 'pull'),
         # Edge case: not cmd argument
         ('/foo/bar?key=tip', 'pull'),
     ])
 def test_get_action(url, expected_action):
     app = simplehg.SimpleHg(application=None,
-                            config={'auth_ret_code': '', 'base_path': ''})
+                            config={'auth_ret_code': '', 'base_path': ''},
+                            registry=None)
     assert expected_action == app._get_action(get_environ(url))
 
 
 @pytest.mark.parametrize(
     'url, expected_repo_name',
     [
         ('/foo?cmd=unbundle&key=tip', 'foo'),
         ('/foo/bar?cmd=pushkey&key=tip', 'foo/bar'),
         ('/foo/bar/baz?cmd=listkeys&key=tip', 'foo/bar/baz'),
         # Repos with trailing slashes.
         ('/foo/?cmd=unbundle&key=tip', 'foo'),
         ('/foo/bar/?cmd=pushkey&key=tip', 'foo/bar'),
         ('/foo/bar/baz/?cmd=listkeys&key=tip', 'foo/bar/baz'),
     ])
 def test_get_repository_name(url, expected_repo_name):
     app = simplehg.SimpleHg(application=None,
-                            config={'auth_ret_code': '', 'base_path': ''})
+                            config={'auth_ret_code': '', 'base_path': ''},
+                            registry=None)
     assert expected_repo_name == app._get_repository_name(get_environ(url))
 
 
 def test_get_config():
     app = simplehg.SimpleHg(application=None,
-                            config={'auth_ret_code': '', 'base_path': ''})
+                            config={'auth_ret_code': '', 'base_path': ''},
+                            registry=None)
     extras = {'foo': 'FOO', 'bar': 'BAR'}
 
     mock_config = Config()
     mock_config.set('a1', 'b1', 'c1')
     mock_config.set('a2', 'b2', 'c2')
     # We mock the call to make_db_config, otherwise we need to wait for the
     # pylonsaspp
     with mock.patch('rhodecode.lib.utils.make_db_config',
                     return_value=mock_config) as make_db_config_mock:
         hg_config = app._create_config(extras, repo_name='test-repo')
 
     make_db_config_mock.assert_called_once_with(repo='test-repo')
     assert isinstance(hg_config, list)
 
     # Remove the entries from the mock_config so to get only the extras
     hg_config.remove(('a1', 'b1', 'c1'))
     hg_config.remove(('a2', 'b2', 'c2'))
 
     assert hg_config[0][:2] == ('rhodecode', 'RC_SCM_DATA')
     assert json.loads(hg_config[0][-1]) == extras
 
 
 def test_create_wsgi_app_uses_scm_app_from_simplevcs():
     config = {
         'auth_ret_code': '',
         'base_path': '',
         'vcs.scm_app_implementation':
             'rhodecode.tests.lib.middleware.mock_scm_app',
     }
-    app = simplehg.SimpleHg(application=None, config=config)
+    app = simplehg.SimpleHg(application=None, config=config, registry=None)
     wsgi_app = app._create_wsgi_app('/tmp/test', 'test_repo', {})
     assert wsgi_app is mock_scm_app.mock_hg_wsgi

rhodecode/tests/lib/middleware/test_simplesvn.py

@@ -1,185 +1,186 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/
 
 from StringIO import StringIO
 
 import pytest
 from mock import patch, Mock
 
 import rhodecode
 from rhodecode.lib.middleware.simplesvn import SimpleSvn, SimpleSvnApp
 
 
 class TestSimpleSvn(object):
     @pytest.fixture(autouse=True)
     def simple_svn(self, pylonsapp):
         self.app = SimpleSvn(
             application='None',
             config={'auth_ret_code': '',
-                    'base_path': rhodecode.CONFIG['base_path']})
+                    'base_path': rhodecode.CONFIG['base_path']},
+            registry=None)
 
     def test_get_config(self):
         extras = {'foo': 'FOO', 'bar': 'BAR'}
         config = self.app._create_config(extras, repo_name='test-repo')
         assert config == extras
 
     @pytest.mark.parametrize(
         'method', ['OPTIONS', 'PROPFIND', 'GET', 'REPORT'])
     def test_get_action_returns_pull(self, method):
         environment = {'REQUEST_METHOD': method}
         action = self.app._get_action(environment)
         assert action == 'pull'
 
     @pytest.mark.parametrize(
         'method', [
             'MKACTIVITY', 'PROPPATCH', 'PUT', 'CHECKOUT', 'MKCOL', 'MOVE',
             'COPY', 'DELETE', 'LOCK', 'UNLOCK', 'MERGE'
         ])
     def test_get_action_returns_push(self, method):
         environment = {'REQUEST_METHOD': method}
         action = self.app._get_action(environment)
         assert action == 'push'
 
     @pytest.mark.parametrize(
         'path, expected_name', [
             ('/hello-svn', 'hello-svn'),
             ('/hello-svn/', 'hello-svn'),
             ('/group/hello-svn/', 'group/hello-svn'),
             ('/group/hello-svn/!svn/vcc/default', 'group/hello-svn'),
         ])
     def test_get_repository_name(self, path, expected_name):
         environment = {'PATH_INFO': path}
         name = self.app._get_repository_name(environment)
         assert name == expected_name
 
     def test_get_repository_name_subfolder(self, backend_svn):
         repo = backend_svn.repo
         environment = {
             'PATH_INFO': '/{}/path/with/subfolders'.format(repo.repo_name)}
         name = self.app._get_repository_name(environment)
         assert name == repo.repo_name
 
     def test_create_wsgi_app(self):
         with patch('rhodecode.lib.middleware.simplesvn.SimpleSvnApp') as (
                 wsgi_app_mock):
             config = Mock()
             wsgi_app = self.app._create_wsgi_app(
                 repo_path='', repo_name='', config=config)
 
         wsgi_app_mock.assert_called_once_with(config)
         assert wsgi_app == wsgi_app_mock()
 
 
 class TestSimpleSvnApp(object):
     data = '<xml></xml>'
     path = '/group/my-repo'
     wsgi_input = StringIO(data)
     environment = {
         'HTTP_DAV': (
             'http://subversion.tigris.org/xmlns/dav/svn/depth,'
             ' http://subversion.tigris.org/xmlns/dav/svn/mergeinfo'),
         'HTTP_USER_AGENT': 'SVN/1.8.11 (x86_64-linux) serf/1.3.8',
         'REQUEST_METHOD': 'OPTIONS',
         'PATH_INFO': path,
         'wsgi.input': wsgi_input,
         'CONTENT_TYPE': 'text/xml',
         'CONTENT_LENGTH': '130'
     }
 
     def setup_method(self, method):
         self.host = 'http://localhost/'
         self.app = SimpleSvnApp(
             config={'subversion_http_server_url': self.host})
 
     def test_get_request_headers_with_content_type(self):
         expected_headers = {
             'Dav': self.environment['HTTP_DAV'],
             'User-Agent': self.environment['HTTP_USER_AGENT'],
             'Content-Type': self.environment['CONTENT_TYPE'],
             'Content-Length': self.environment['CONTENT_LENGTH']
         }
         headers = self.app._get_request_headers(self.environment)
         assert headers == expected_headers
 
     def test_get_request_headers_without_content_type(self):
         environment = self.environment.copy()
         environment.pop('CONTENT_TYPE')
         expected_headers = {
             'Dav': environment['HTTP_DAV'],
             'Content-Length': self.environment['CONTENT_LENGTH'],
             'User-Agent': environment['HTTP_USER_AGENT'],
         }
         request_headers = self.app._get_request_headers(environment)
         assert request_headers == expected_headers
 
     def test_get_response_headers(self):
         headers = {
             'Connection': 'keep-alive',
             'Keep-Alive': 'timeout=5, max=100',
             'Transfer-Encoding': 'chunked',
             'Content-Encoding': 'gzip',
             'MS-Author-Via': 'DAV',
             'SVN-Supported-Posts': 'create-txn-with-props'
         }
         expected_headers = [
             ('MS-Author-Via', 'DAV'),
             ('SVN-Supported-Posts', 'create-txn-with-props')
         ]
         response_headers = self.app._get_response_headers(headers)
         assert sorted(response_headers) == sorted(expected_headers)
 
     def test_get_url(self):
         url = self.app._get_url(self.path)
         expected_url = '{}{}'.format(self.host.strip('/'), self.path)
         assert url == expected_url
 
     def test_call(self):
         start_response = Mock()
         response_mock = Mock()
         response_mock.headers = {
             'Content-Encoding': 'gzip',
             'MS-Author-Via': 'DAV',
             'SVN-Supported-Posts': 'create-txn-with-props'
         }
         response_mock.status_code = 200
         response_mock.reason = 'OK'
         with patch('rhodecode.lib.middleware.simplesvn.requests.request') as (
                 request_mock):
             request_mock.return_value = response_mock
             self.app(self.environment, start_response)
 
         expected_url = '{}{}'.format(self.host.strip('/'), self.path)
         expected_request_headers = {
             'Dav': self.environment['HTTP_DAV'],
             'User-Agent': self.environment['HTTP_USER_AGENT'],
             'Content-Type': self.environment['CONTENT_TYPE'],
             'Content-Length': self.environment['CONTENT_LENGTH']
         }
         expected_response_headers = [
             ('SVN-Supported-Posts', 'create-txn-with-props'),
             ('MS-Author-Via', 'DAV')
         ]
         request_mock.assert_called_once_with(
             self.environment['REQUEST_METHOD'], expected_url,
             data=self.data, headers=expected_request_headers)
         response_mock.iter_content.assert_called_once_with(chunk_size=1024)
         args, _ = start_response.call_args
         assert args[0] == '200 OK'
         assert sorted(args[1]) == sorted(expected_response_headers)

rhodecode/tests/lib/middleware/test_simplevcs.py

@@ -1,304 +1,305 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 base64
 
 import mock
 import pytest
 import webtest.app
 
 from rhodecode.lib.caching_query import FromCache
-from rhodecode.lib.hooks_daemon import (
-    Pyro4HooksCallbackDaemon, DummyHooksCallbackDaemon,
-    HttpHooksCallbackDaemon)
+from rhodecode.lib.hooks_daemon import DummyHooksCallbackDaemon
 from rhodecode.lib.middleware import simplevcs
 from rhodecode.lib.middleware.https_fixup import HttpsFixup
 from rhodecode.lib.middleware.utils import scm_app
 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
 from rhodecode.tests.utils import set_anonymous_access
 
 
 class StubVCSController(simplevcs.SimpleVCS):
 
     SCM = 'hg'
     stub_response_body = tuple()
 
     def _get_repository_name(self, environ):
         return HG_REPO
 
     def _get_action(self, environ):
         return "pull"
 
     def _create_wsgi_app(self, repo_path, repo_name, config):
         def fake_app(environ, start_response):
             start_response('200 OK', [])
             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')
 
     set_anonymous_access(True)
-    controller = StubVCSController(pylonsapp, pylonsapp.config)
+    controller = StubVCSController(pylonsapp, pylonsapp.config, None)
     app = HttpsFixup(controller, pylonsapp.config)
     app = webtest.app.TestApp(app)
 
     _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)
     query = query.options(FromCache(
         "sql_cache_short", "get_user_%s" % _hash_key(user.username)))
     query.invalidate()
     Session().expire(user)
 
 
 @pytest.fixture
 def disable_anonymous_user(request, pylonsapp):
     set_anonymous_access(False)
 
     @request.addfinalizer
     def cleanup():
         set_anonymous_access(True)
 
 
 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):
-    controller = StubFailVCSController(pylonsapp, pylonsapp.config)
+    controller = StubFailVCSController(pylonsapp, pylonsapp.config, None)
     controller = HttpsFixup(controller, pylonsapp.config)
     controller = webtest.app.TestApp(controller)
     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):
-    controller = StubVCSController(pylonsapp, pylonsapp.config)
+    controller = StubVCSController(pylonsapp, pylonsapp.config, None)
     assert controller.scm_app is scm_app
 
 
 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')
-    controller = StubVCSController(pylonsapp, config)
+    controller = StubVCSController(pylonsapp, config, None)
     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
 
 
 @mock.patch.multiple(
     'Pyro4.config', SERVERTYPE='multiplex', POLLTIMEOUT=0.01)
 class TestGenerateVcsResponse:
 
     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):
-        controller = StubVCSController(None, {'base_path': 'fake_base_path'})
+        settings = {
+            'base_path': 'fake_base_path',
+            'vcs.hooks.protocol': 'http',
+            'vcs.hooks.direct_calls': False,
+        }
+        controller = StubVCSController(None, settings, None)
         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',
             repo_name='fake_repo_name',
             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
 
 
 class TestInitializeGenerator:
 
     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):
-    def test_calls_imported_prepare_callback_daemon(self):
-        config = {
-            'base_path': 'fake_base_path',
-            'vcs.hooks.direct_calls': False,
-            'vcs.hooks.protocol': 'http'
-        }
+    def test_calls_imported_prepare_callback_daemon(self, pylonsapp):
+        settings = pylonsapp.application.config
         expected_extras = {'extra1': 'value1'}
         daemon = DummyHooksCallbackDaemon()
 
-        controller = StubVCSController(None, config)
+        controller = StubVCSController(None, settings, None)
         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(
-            expected_extras, protocol='http', use_direct_calls=False)
+            expected_extras,
+            protocol=settings['vcs.hooks.protocol'],
+            use_direct_calls=settings['vcs.hooks.direct_calls'])
 
         assert callback_daemon == daemon
         assert extras == extras

rhodecode/tests/lib/middleware/test_vcs.py

@@ -1,134 +1,136 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/
 
 from mock import patch, Mock
 
 import rhodecode
 from rhodecode.lib.middleware import vcs
 
 
 def test_is_hg():
     environ = {
         'PATH_INFO': '/rhodecode-dev',
         'QUERY_STRING': 'cmd=changegroup',
         'HTTP_ACCEPT': 'application/mercurial'
     }
     assert vcs.is_hg(environ)
 
 
 def test_is_hg_no_cmd():
     environ = {
         'PATH_INFO': '/rhodecode-dev',
         'QUERY_STRING': '',
         'HTTP_ACCEPT': 'application/mercurial'
     }
     assert not vcs.is_hg(environ)
 
 
 def test_is_hg_empty_cmd():
     environ = {
         'PATH_INFO': '/rhodecode-dev',
         'QUERY_STRING': 'cmd=',
         'HTTP_ACCEPT': 'application/mercurial'
     }
     assert not vcs.is_hg(environ)
 
 
 def test_is_svn_returns_true_if_subversion_is_in_a_dav_header():
     environ = {
         'PATH_INFO': '/rhodecode-dev',
         'HTTP_DAV': 'http://subversion.tigris.org/xmlns/dav/svn/log-revprops'
     }
     assert vcs.is_svn(environ) is True
 
 
 def test_is_svn_returns_false_if_subversion_is_not_in_a_dav_header():
     environ = {
         'PATH_INFO': '/rhodecode-dev',
         'HTTP_DAV': 'http://stuff.tigris.org/xmlns/dav/svn/log-revprops'
     }
     assert vcs.is_svn(environ) is False
 
 
 def test_is_svn_returns_false_if_no_dav_header():
     environ = {
         'PATH_INFO': '/rhodecode-dev',
     }
     assert vcs.is_svn(environ) is False
 
 
 def test_is_svn_returns_true_if_magic_path_segment():
     environ = {
         'PATH_INFO': '/stub-repository/!svn/rev/4',
     }
     assert vcs.is_svn(environ)
 
 
 def test_is_svn_allows_to_configure_the_magic_path(monkeypatch):
     """
     This is intended as a fallback in case someone has configured his
     Subversion server with a different magic path segment.
     """
     monkeypatch.setitem(
         rhodecode.CONFIG, 'rhodecode_subversion_magic_path', '/!my-magic')
     environ = {
         'PATH_INFO': '/stub-repository/!my-magic/rev/4',
     }
     assert vcs.is_svn(environ)
 
 
 class TestVCSMiddleware(object):
     def test_get_handler_app_retuns_svn_app_when_proxy_enabled(self):
         environ = {
             'PATH_INFO': 'rhodecode-dev',
             'HTTP_DAV': 'http://subversion.tigris.org/xmlns/dav/svn/log'
         }
         app = Mock()
         config = Mock()
+        registry = Mock()
         middleware = vcs.VCSMiddleware(
-            app, config=config, appenlight_client=None)
+            app, config=config, appenlight_client=None, registry=registry)
         snv_patch = patch('rhodecode.lib.middleware.vcs.SimpleSvn')
         settings_patch = patch.dict(
             rhodecode.CONFIG,
             {'rhodecode_proxy_subversion_http_requests': True})
         with snv_patch as svn_mock, settings_patch:
             svn_mock.return_value = None
             middleware._get_handler_app(environ)
 
-        svn_mock.assert_called_once_with(app, config)
+        svn_mock.assert_called_once_with(app, config, registry)
 
     def test_get_handler_app_retuns_no_svn_app_when_proxy_disabled(self):
         environ = {
             'PATH_INFO': 'rhodecode-dev',
             'HTTP_DAV': 'http://subversion.tigris.org/xmlns/dav/svn/log'
         }
         app = Mock()
         config = Mock()
+        registry = Mock()
         middleware = vcs.VCSMiddleware(
-            app, config=config, appenlight_client=None)
+            app, config=config, appenlight_client=None, registry=registry)
         snv_patch = patch('rhodecode.lib.middleware.vcs.SimpleSvn')
         settings_patch = patch.dict(
             rhodecode.CONFIG,
             {'rhodecode_proxy_subversion_http_requests': False})
         with snv_patch as svn_mock, settings_patch:
             app = middleware._get_handler_app(environ)
 
         assert svn_mock.call_count == 0
         assert app is None