rhodecode/tests/vcs_operations/test_vcs_operations_svn.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/

"""
Test suite for making push/pull operations, on specially modified INI files

.. important::

   You must have git >= 1.8.5 for tests to work fine. With 68b939b git started
   to redirect things to stderr instead of stdout.
"""


import time
import pytest

from rhodecode.model.db import Repository, UserIpMap
from rhodecode.model.meta import Session
from rhodecode.model.repo import RepoModel
from rhodecode.model.user import UserModel
from rhodecode.tests import (SVN_REPO, TEST_USER_ADMIN_LOGIN)


from rhodecode.tests.vcs_operations import (
    Command, _check_proper_clone, _check_proper_svn_push,
    _add_files_and_push, SVN_REPO_WITH_GROUP)


def get_cli_flags(username, password):
    flags = '--no-auth-cache --non-interactive'
    auth = ''
    if username and password:
        auth = f'--username {username} --password {password}'
    return flags, auth


@pytest.mark.usefixtures("disable_locking", "disable_anonymous_user")
class TestVCSOperations(object):

    def test_clone_svn_repo_by_admin(self, rc_web_server, tmpdir):
        clone_url = rc_web_server.repo_clone_url(SVN_REPO)
        username, password = rc_web_server.repo_clone_credentials()

        cmd = Command('/tmp')

        flags, auth = get_cli_flags(username, password)

        stdout, stderr = Command('/tmp').execute(
            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)

        _check_proper_clone(stdout, stderr, 'svn')
        cmd.assert_returncode_success()

    def test_clone_svn_repo_by_id_by_admin(self, rc_web_server, tmpdir):
        repo_id = Repository.get_by_repo_name(SVN_REPO).repo_id
        username, password = rc_web_server.repo_clone_credentials()

        clone_url = rc_web_server.repo_clone_url('_%s' % repo_id)
        cmd = Command('/tmp')

        flags, auth = get_cli_flags(username, password)

        stdout, stderr = Command('/tmp').execute(
            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)

        _check_proper_clone(stdout, stderr, 'svn')
        cmd.assert_returncode_success()

    def test_clone_svn_repo_with_group_by_admin(self, rc_web_server, tmpdir):
        clone_url = rc_web_server.repo_clone_url(SVN_REPO_WITH_GROUP)
        username, password = rc_web_server.repo_clone_credentials()

        flags, auth = get_cli_flags(username, password)

        stdout, stderr = Command('/tmp').execute(
            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)

        _check_proper_clone(stdout, stderr, 'svn')
        cmd.assert_returncode_success()

    def test_clone_wrong_credentials_svn(self, rc_web_server, tmpdir):
        clone_url = rc_web_server.repo_clone_url(SVN_REPO)
        username, password = rc_web_server.repo_clone_credentials()
        password = 'bad-password'

        flags, auth = get_cli_flags(username, password)

        stdout, stderr = Command('/tmp').execute(
            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
        assert 'fatal: Authentication failed' in stderr

    def test_clone_svn_with_slashes(self, rc_web_server, tmpdir):
        clone_url = rc_web_server.repo_clone_url('//' + SVN_REPO)
        username, password = '', ''
        flags, auth = get_cli_flags(username, password)

        stdout, stderr = Command('/tmp').execute(
            f'svn checkout {flags} {auth}', clone_url)

        assert 'not found' in stderr

    def test_clone_existing_path_svn_not_in_database(
            self, rc_web_server, tmpdir, fs_repo_only):
        db_name = fs_repo_only('not-in-db-git', repo_type='git')
        clone_url = rc_web_server.repo_clone_url(db_name)
        username, password = '', ''
        flags, auth = get_cli_flags(username, password)

        stdout, stderr = Command('/tmp').execute(
            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
        assert 'not found' in stderr

    def test_clone_existing_path_svn_not_in_database_different_scm(
            self, rc_web_server, tmpdir, fs_repo_only):
        db_name = fs_repo_only('not-in-db-hg', repo_type='hg')
        clone_url = rc_web_server.repo_clone_url(db_name)

        username, password = '', ''
        flags, auth = get_cli_flags(username, password)

        stdout, stderr = Command('/tmp').execute(
            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
        assert 'not found' in stderr

    def test_clone_non_existing_store_path_svn(self, rc_web_server, tmpdir, user_util):
        repo = user_util.create_repo(repo_type='git')
        clone_url = rc_web_server.repo_clone_url(repo.repo_name)

        # Damage repo by removing it's folder
        RepoModel()._delete_filesystem_repo(repo)

        username, password = '', ''
        flags, auth = get_cli_flags(username, password)

        stdout, stderr = Command('/tmp').execute(
            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
        assert 'not found' in stderr

    def test_push_new_file_svn(self, rc_web_server, tmpdir):
        clone_url = rc_web_server.repo_clone_url(SVN_REPO)
        username, password = '', ''
        flags, auth = get_cli_flags(username, password)

        stdout, stderr = Command('/tmp').execute(
            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)

        # commit some stuff into this repo
        stdout, stderr = _add_files_and_push(
            'svn', tmpdir.strpath, clone_url=clone_url, username=username, password=password)

        _check_proper_svn_push(stdout, stderr)

    def test_push_wrong_credentials_svn(self, rc_web_server, tmpdir):
        clone_url = rc_web_server.repo_clone_url(SVN_REPO)

        username, password = rc_web_server.repo_clone_credentials()
        flags, auth = get_cli_flags(username, password)

        stdout, stderr = Command('/tmp').execute(
            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)

        push_url = rc_web_server.repo_clone_url(
            SVN_REPO, user='bad', passwd='name')
        stdout, stderr = _add_files_and_push(
            'svn', tmpdir.strpath, clone_url=push_url, username=username, password=password)

        assert 'fatal: Authentication failed' in stderr

    def test_push_back_to_wrong_url_svn(self, rc_web_server, tmpdir):
        clone_url = rc_web_server.repo_clone_url(SVN_REPO)
        username, password = '', ''
        flags, auth = get_cli_flags(username, password)

        stdout, stderr = Command('/tmp').execute(
            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)

        stdout, stderr = _add_files_and_push(
            'svn', tmpdir.strpath,
            clone_url=rc_web_server.repo_clone_url('not-existing'), username=username, password=password)

        assert 'not found' in stderr

    def test_ip_restriction_svn(self, rc_web_server, tmpdir):
        user_model = UserModel()
        username, password = '', ''
        flags, auth = get_cli_flags(username, password)

        try:
            user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')
            Session().commit()
            time.sleep(2)
            clone_url = rc_web_server.repo_clone_url(SVN_REPO)

            stdout, stderr = Command('/tmp').execute(
                f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
            msg = "The requested URL returned error: 403"
            assert msg in stderr
        finally:
            # release IP restrictions
            for ip in UserIpMap.getAll():
                UserIpMap.delete(ip.ip_id)
            Session().commit()

        time.sleep(2)

        cmd = Command('/tmp')
        stdout, stderr = cmd.execute(f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
        cmd.assert_returncode_success()
        _check_proper_clone(stdout, stderr, 'svn')