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