# -*- coding: utf-8 -*- # Copyright (C) 2010-2020 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 . # # 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 from rhodecode.lib.middleware.simplesvn import SimpleSvn, SimpleSvnApp from rhodecode.lib.utils import get_rhodecode_base_path class TestSimpleSvn(object): @pytest.fixture(autouse=True) def simple_svn(self, baseapp, request_stub): base_path = get_rhodecode_base_path() self.app = SimpleSvn( config={'auth_ret_code': '', 'base_path': base_path}, registry=request_stub.registry) 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.object(SimpleSvn, '_is_svn_enabled') as mock_method: mock_method.return_value = False with patch('rhodecode.lib.middleware.simplesvn.DisabledSimpleSvnApp') 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() def test_create_wsgi_app_when_enabled(self): with patch.object(SimpleSvn, '_is_svn_enabled') as mock_method: mock_method.return_value = True 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 = '' 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/' base_path = get_rhodecode_base_path() self.app = SimpleSvnApp( config={'subversion_http_server_url': self.host, 'base_path': base_path}) 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) @pytest.mark.parametrize('svn_http_url, path_info, expected_url', [ ('http://localhost:8200', '/repo_name', 'http://localhost:8200/repo_name'), ('http://localhost:8200///', '/repo_name', 'http://localhost:8200/repo_name'), ('http://localhost:8200', '/group/repo_name', 'http://localhost:8200/group/repo_name'), ('http://localhost:8200/', '/group/repo_name', 'http://localhost:8200/group/repo_name'), ('http://localhost:8200/prefix', '/repo_name', 'http://localhost:8200/prefix/repo_name'), ('http://localhost:8200/prefix', 'repo_name', 'http://localhost:8200/prefix/repo_name'), ('http://localhost:8200/prefix', '/group/repo_name', 'http://localhost:8200/prefix/group/repo_name') ]) def test_get_url(self, svn_http_url, path_info, expected_url): url = self.app._get_url(svn_http_url, path_info) 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, stream=False) 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)