rhodecode/config/utils.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 os
import platform

from rhodecode.lib.type_utils import str2bool

DEFAULT_USER = 'default'


def configure_vcs(config):
    """
    Patch VCS config with some RhodeCode specific stuff
    """
    from rhodecode.lib.vcs import conf
    import rhodecode.lib.vcs.conf.settings

    conf.settings.BACKENDS = {
        'hg': 'rhodecode.lib.vcs.backends.hg.MercurialRepository',
        'git': 'rhodecode.lib.vcs.backends.git.GitRepository',
        'svn': 'rhodecode.lib.vcs.backends.svn.SubversionRepository',
    }

    conf.settings.HOOKS_PROTOCOL = config['vcs.hooks.protocol.v2']
    conf.settings.HOOKS_HOST = config['vcs.hooks.host']
    conf.settings.ALIASES[:] = config['vcs.backends']
    conf.settings.SVN_COMPATIBLE_VERSION = config['vcs.svn.compatible_version']


def initialize_database(config):
    from rhodecode.lib.utils2 import engine_from_config, get_encryption_key
    from rhodecode.model import init_model
    engine = engine_from_config(config, 'sqlalchemy.db1.')
    init_model(engine, encryption_key=get_encryption_key(config))

def initialize_test_environment(settings):
    skip_test_env = str2bool(os.environ.get('RC_NO_TEST_ENV'))
    if skip_test_env:
        return

    repo_store_path = os.environ.get('RC_TEST_ENV_REPO_STORE') or settings['repo_store.path']

    from rhodecode.lib.utils import (
        create_test_directory, create_test_database, create_test_repositories,
        create_test_index)

    create_test_directory(repo_store_path)

    create_test_database(repo_store_path, settings)
    # test repos
    create_test_repositories(repo_store_path, settings)
    create_test_index(repo_store_path, settings)


def get_vcs_server_protocol(config):
    return config['vcs.server.protocol']


def set_instance_id(config):
    """
    Sets a dynamic generated config['instance_id'] if missing or '*'
    E.g instance_id = *cluster-1 or instance_id = *
    """

    config['instance_id'] = config.get('instance_id') or ''
    instance_id = config['instance_id']
    if instance_id.startswith('*') or not instance_id:
        prefix = instance_id.lstrip('*')
        _platform_id = platform.uname()[1] or 'instance'
        config['instance_id'] = '{prefix}uname:{platform}-pid:{pid}'.format(
            prefix=prefix,
            platform=_platform_id,
            pid=os.getpid())


def get_default_user_id():
    from sqlalchemy import text
    from rhodecode.model import meta

    engine = meta.get_engine()
    with meta.SA_Session(engine) as session:
        result = session.execute(text(
            "SELECT user_id from users where username = :uname"
        ), {'uname': DEFAULT_USER})
        user = result.first()
        if not user:
            raise ValueError('Unable to retrieve default user data from DB')
        user_id = user[0]

    return user_id