rhodecode/tests/fixtures/fixture_pyramid.py

# Copyright (C) 2010-2024 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.fixtures.rc_fixture import TestINI
from rhodecode.tests import TESTS_TMP_PATH
from rhodecode.tests.server_utils import RcVCSServer
from rhodecode.tests.server_utils import RcWebServer
from rhodecode.tests.server_utils import CeleryServer


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

    def factory(request, store_dir, overrides=(), config_file=None, port=None, log_file=None, workers="3", env=None, info_prefix=""):
        env = env or {"RC_NO_TEST_ENV": "1"}
        vcsserver_port = port
        if port is None:
            vcsserver_port = get_available_port()

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

        if getattr(request, 'param', None):
            config_overrides = [request.param]
            overrides.extend(config_overrides)

        option_name = "vcsserver_config"
        override_option_name = None
        if not config_file:
            config_file = get_config(
                request.config,
                option_name=option_name,
                override_option_name=override_option_name,
                overrides=overrides,
                basetemp=store_dir,
                prefix=f"{info_prefix}test_vcsserver_ini_",
            )
        server = RcVCSServer(config_file, log_file, workers, env=env, info_prefix=info_prefix)
        server.start()

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

        server.wait_until_ready()
        return server

    return factory


@pytest.fixture(scope="session")
def rhodecode_factory():
    def factory(request, store_dir, overrides=(), config_file=None, port=None, log_file=None, workers="3", env=None, info_prefix=""):
        env = env or {"RC_NO_TEST_ENV": "1"}
        rhodecode_port = port
        if port is None:
            rhodecode_port = get_available_port()

        overrides = list(overrides)
        overrides.append({"server:main": {"port": rhodecode_port}})
        overrides.append({"app:main": {"use_celery": "true"}})
        overrides.append({"app:main": {"celery.task_always_eager": "false"}})

        if getattr(request, 'param', None):
            config_overrides = [request.param]
            overrides.extend(config_overrides)


        option_name = "rhodecode_config"
        override_option_name = None
        if not config_file:
            config_file = get_config(
                request.config,
                option_name=option_name,
                override_option_name=override_option_name,
                overrides=overrides,
                basetemp=store_dir,
                prefix=f"{info_prefix}test_rhodecode_ini",
            )

        server = RcWebServer(config_file, log_file, workers, env, info_prefix=info_prefix)
        server.start()

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

        server.wait_until_ready()
        return server

    return factory


@pytest.fixture(scope="session")
def celery_factory():
    def factory(request, store_dir, overrides=(), config_file=None, port=None, log_file=None, workers="3", env=None, info_prefix=""):
        env = env or {"RC_NO_TEST_ENV": "1"}
        rhodecode_port = port

        overrides = list(overrides)
        overrides.append({"app:main": {"use_celery": "true"}})
        overrides.append({"app:main": {"celery.task_always_eager": "false"}})
        config_overrides = None

        if getattr(request, 'param', None):
            config_overrides = [request.param]
            overrides.extend(config_overrides)

        option_name = "celery_config"
        override_option_name = None

        if not config_file:
            config_file = get_config(
                request.config,
                option_name=option_name,
                override_option_name=override_option_name,
                overrides=overrides,
                basetemp=store_dir,
                prefix=f"{info_prefix}test_celery_ini_",
            )

        server = CeleryServer(config_file, log_file, workers, env, info_prefix=info_prefix)
        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()


def _ini_config_factory(request, base_dir, 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=f"{option_name}_override",
        overrides=overrides,
        basetemp=base_dir,
        prefix="test_rce_",
    )
    return filename


@pytest.fixture(scope="session")
def ini_config(request, tmpdir_factory, rcserver_port, vcsserver_port):
    base_dir = tmpdir_factory.getbasetemp().strpath
    return _ini_config_factory(request, base_dir, rcserver_port, vcsserver_port)


@pytest.fixture(scope="session")
def ini_config_factory(request, tmpdir_factory, rcserver_port, vcsserver_port):
    def _factory(ini_config_basedir, overrides=()):
        return _ini_config_factory(request, ini_config_basedir, rcserver_port, vcsserver_port)

    return _factory


@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()
    return port


@pytest.fixture(scope="session")
def vcsserver_port(request):
    port = request.config.getoption("--vcsserver-port")
    if port is None:
        port = get_available_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, ini_store_dir, template_ini):
        self._ini_store_dir = ini_store_dir
        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._ini_store_dir
        )
        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`.
    """
    try:
        config_file = config.getoption(option_name)
    except ValueError:
        config_file = None

    if not config_file:
        config_file = config.getini(option_name)

    if not config_file:
        pytest.exit(f"Configuration error, could not extract {option_name}.")

    overrides = overrides or []
    if override_option_name:
        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()