rhodecode/tests/lib/middleware/test_simplesvn.py

# Copyright (C) 2010-2023 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 io
from base64 import b64encode

import pytest
from unittest.mock import patch, Mock, MagicMock

from rhodecode.lib.middleware.simplesvn import SimpleSvn, SimpleSvnApp
from rhodecode.lib.utils import get_rhodecode_base_path
from rhodecode.tests import SVN_REPO, TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS


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()


def basic_auth(username, password):
    token = b64encode(f"{username}:{password}".encode('utf-8')).decode("ascii")
    return f'Basic {token}'


class TestSimpleSvnApp(object):
    data = b'<xml></xml>'
    path = SVN_REPO
    wsgi_input = io.BytesIO(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.14.1 (x86_64-linux) serf/1.3.8',
        'REQUEST_METHOD': 'OPTIONS',
        'PATH_INFO': path,
        'wsgi.input': wsgi_input,
        'CONTENT_TYPE': 'text/xml',
        'CONTENT_LENGTH': '130',
        'Authorization': basic_auth(TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS)
    }

    def setup_method(self, method):
        # note(marcink): this is hostname from docker compose used for testing...
        self.host = 'http://svn:8090'
        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'],
            'Authorization': self.environment['Authorization']
        }
        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'],
            'Authorization': self.environment['Authorization']
        }
        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'
        }

        from rhodecode.lib.middleware.simplesvn import requests
        original_request = requests.Session.request

        with patch('rhodecode.lib.middleware.simplesvn.requests.Session.request', autospec=True) as request_mock:
            # Use side_effect to call the original method
            request_mock.side_effect = original_request
            self.app(self.environment, start_response)

        expected_url = f'{self.host.strip("/")}/{self.path}'
        expected_request_headers = {
            'Dav': self.environment['HTTP_DAV'],
            'User-Agent': self.environment['HTTP_USER_AGENT'],
            'Authorization': self.environment['Authorization'],
            'Content-Type': self.environment['CONTENT_TYPE'],
            'Content-Length': self.environment['CONTENT_LENGTH'],
        }

        # Check if the method was called
        assert request_mock.called
        assert request_mock.call_count == 1

        # Extract the session instance from the first call
        called_with_session = request_mock.call_args[0][0]

        request_mock.assert_called_once_with(
            called_with_session,
            self.environment['REQUEST_METHOD'], expected_url,
            data=self.data, headers=expected_request_headers, stream=False)

        expected_response_headers = [
            ('SVN-Supported-Posts', 'create-txn-with-props'),
            ('MS-Author-Via', 'DAV'),
        ]

        # TODO: the svn doesn't have a repo for testing
        #args, _ = start_response.call_args
        #assert args[0] == '200 OK'
        #assert sorted(args[1]) == sorted(expected_response_headers)