rhodecode/tests/fixture_mods/fixture_pyramid.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 pytest

from rhodecode.lib.config_utils import get_app_config
from rhodecode.tests.fixture import TestINI
from rhodecode.tests import TESTS_TMP_PATH
from rhodecode.tests.server_utils import RcVCSServer


@pytest.fixture(scope='session')
def vcsserver(request, vcsserver_port, vcsserver_factory):
    """
    Session scope VCSServer.

    Tests which need the VCSServer have to rely on this fixture in order
    to ensure it will be running.

    For specific needs, the fixture vcsserver_factory can be used. It allows to
    adjust the configuration file for the test run.

    Command line args:

    --without-vcsserver: Allows to switch this fixture off. You have to
    manually start the server.

    --vcsserver-port: Will expect the VCSServer to listen on this port.
    """

    if not request.config.getoption('with_vcsserver'):
        return None

    return vcsserver_factory(
        request, vcsserver_port=vcsserver_port)


@pytest.fixture(scope='session')
def vcsserver_factory(tmpdir_factory):
    """
    Use this if you need a running vcsserver with a special configuration.
    """

    def factory(request, overrides=(), vcsserver_port=None,
                log_file=None, workers='3'):

        if vcsserver_port is None:
            vcsserver_port = get_available_port()

        overrides = list(overrides)
        overrides.append({'server:main': {'port': vcsserver_port}})

        option_name = 'vcsserver_config_http'
        override_option_name = 'vcsserver_config_override'
        config_file = get_config(
            request.config, option_name=option_name,
            override_option_name=override_option_name, overrides=overrides,
            basetemp=tmpdir_factory.getbasetemp().strpath,
            prefix='test_vcs_')

        server = RcVCSServer(config_file, log_file, workers)
        server.start()

        @request.addfinalizer
        def cleanup():
            server.shutdown()

        server.wait_until_ready()
        return server

    return factory


def _use_log_level(config):
    level = config.getoption('test_loglevel') or 'critical'
    return level.upper()


@pytest.fixture(scope='session')
def ini_config(request, tmpdir_factory, rcserver_port, vcsserver_port):
    option_name = 'pyramid_config'
    log_level = _use_log_level(request.config)

    overrides = [
        {'server:main': {'port': rcserver_port}},
        {'app:main': {
            'cache_dir': '%(here)s/rc-tests/rc_data',
            'vcs.server': f'localhost:{vcsserver_port}',
            # johbo: We will always start the VCSServer on our own based on the
            # fixtures of the test cases. For the test run it must always be
            # off in the INI file.
            'vcs.start_server': 'false',

            'vcs.server.protocol': 'http',
            'vcs.scm_app_implementation': 'http',
            'vcs.svn.proxy.enabled': 'true',
            'vcs.hooks.protocol.v2': 'celery',
            'vcs.hooks.host': '*',
            'repo_store.path': TESTS_TMP_PATH,
            'app.service_api.token': 'service_secret_token',
        }},

        {'handler_console': {
            'class': 'StreamHandler',
            'args': '(sys.stderr,)',
            'level': log_level,
        }},

    ]

    filename = get_config(
        request.config, option_name=option_name,
        override_option_name='{}_override'.format(option_name),
        overrides=overrides,
        basetemp=tmpdir_factory.getbasetemp().strpath,
        prefix='test_rce_')
    return filename


@pytest.fixture(scope='session')
def ini_settings(ini_config):
    ini_path = ini_config
    return get_app_config(ini_path)


def get_available_port(min_port=40000, max_port=55555):
    from rhodecode.lib.utils2 import get_available_port as _get_port
    return _get_port(min_port, max_port)


@pytest.fixture(scope='session')
def rcserver_port(request):
    port = get_available_port()
    print(f'Using rhodecode port {port}')
    return port


@pytest.fixture(scope='session')
def vcsserver_port(request):
    port = request.config.getoption('--vcsserver-port')
    if port is None:
        port = get_available_port()
        print(f'Using vcsserver port {port}')
    return port


@pytest.fixture(scope='session')
def available_port_factory() -> get_available_port:
    """
    Returns a callable which returns free port numbers.
    """
    return get_available_port


@pytest.fixture()
def available_port(available_port_factory):
    """
    Gives you one free port for the current test.

    Uses "available_port_factory" to retrieve the port.
    """
    return available_port_factory()


@pytest.fixture(scope='session')
def testini_factory(tmpdir_factory, ini_config):
    """
    Factory to create an INI file based on TestINI.

    It will make sure to place the INI file in the correct directory.
    """
    basetemp = tmpdir_factory.getbasetemp().strpath
    return TestIniFactory(basetemp, ini_config)


class TestIniFactory(object):

    def __init__(self, basetemp, template_ini):
        self._basetemp = basetemp
        self._template_ini = template_ini

    def __call__(self, ini_params, new_file_prefix='test'):
        ini_file = TestINI(
            self._template_ini, ini_params=ini_params,
            new_file_prefix=new_file_prefix, dir=self._basetemp)
        result = ini_file.create()
        return result


def get_config(
        config, option_name, override_option_name, overrides=None,
        basetemp=None, prefix='test'):
    """
    Find a configuration file and apply overrides for the given `prefix`.
    """
    config_file = (
        config.getoption(option_name) or config.getini(option_name))
    if not config_file:
        pytest.exit(
            "Configuration error, could not extract {}.".format(option_name))

    overrides = overrides or []
    config_override = config.getoption(override_option_name)
    if config_override:
        overrides.append(config_override)
    temp_ini_file = TestINI(
        config_file, ini_params=overrides, new_file_prefix=prefix,
        dir=basetemp)

    return temp_ini_file.create()