# HG changeset patch # User RhodeCode Admin # Date 2024-02-28 08:30:58 # Node ID 359b5cacb371f930bdc1b4a99424e4cef15d0f16 # Parent d7dccd8eb2f39d58b1fb5a29288306b4dabac2a5 feat(ssh-wrapper-speedup): major rewrite of code to address imports problem with ssh-wrapper-v2 - use bootstrapped settings rather than config - use more code split to make sure we don't import heavy code diff --git a/rhodecode/apps/ssh_support/lib/backends/__init__.py b/rhodecode/apps/ssh_support/lib/backends/__init__.py --- a/rhodecode/apps/ssh_support/lib/backends/__init__.py +++ b/rhodecode/apps/ssh_support/lib/backends/__init__.py @@ -20,12 +20,11 @@ import os import re import logging import datetime -import configparser from sqlalchemy import Table -from rhodecode.lib.utils import call_service_api +from rhodecode.lib.api_utils import call_service_api from rhodecode.lib.utils2 import AttributeDict -from rhodecode.model.scm import ScmModel +from rhodecode.lib.vcs.exceptions import ImproperlyConfiguredError from .hg import MercurialServer from .git import GitServer @@ -39,7 +38,7 @@ class SshWrapper(object): svn_cmd_pat = re.compile(r'^svnserve -t') def __init__(self, command, connection_info, mode, - user, user_id, key_id: int, shell, ini_path: str, env): + user, user_id, key_id: int, shell, ini_path: str, settings, env): self.command = command self.connection_info = connection_info self.mode = mode @@ -49,15 +48,9 @@ class SshWrapper(object): self.shell = shell self.ini_path = ini_path self.env = env - - self.config = self.parse_config(ini_path) + self.settings = settings self.server_impl = None - def parse_config(self, config_path): - parser = configparser.ConfigParser() - parser.read(config_path) - return parser - def update_key_access_time(self, key_id): from rhodecode.model.meta import raw_query_executor, Base @@ -162,6 +155,9 @@ class SshWrapper(object): return vcs_type, repo_name, mode def serve(self, vcs, repo, mode, user, permissions, branch_permissions): + # TODO: remove this once we have .ini defined access path... + from rhodecode.model.scm import ScmModel + store = ScmModel().repos_path check_branch_perms = False @@ -186,7 +182,7 @@ class SshWrapper(object): server = MercurialServer( store=store, ini_path=self.ini_path, repo_name=repo, user=user, - user_permissions=permissions, config=self.config, env=self.env) + user_permissions=permissions, settings=self.settings, env=self.env) self.server_impl = server return server.run(tunnel_extras=extras) @@ -194,7 +190,7 @@ class SshWrapper(object): server = GitServer( store=store, ini_path=self.ini_path, repo_name=repo, repo_mode=mode, user=user, - user_permissions=permissions, config=self.config, env=self.env) + user_permissions=permissions, settings=self.settings, env=self.env) self.server_impl = server return server.run(tunnel_extras=extras) @@ -202,7 +198,7 @@ class SshWrapper(object): server = SubversionServer( store=store, ini_path=self.ini_path, repo_name=None, user=user, - user_permissions=permissions, config=self.config, env=self.env) + user_permissions=permissions, settings=self.settings, env=self.env) self.server_impl = server return server.run(tunnel_extras=extras) @@ -269,6 +265,35 @@ class SshWrapperStandalone(SshWrapper): New version of SshWrapper designed to be depended only on service API """ repos_path = None + service_api_host: str + service_api_token: str + api_url: str + + def __init__(self, command, connection_info, mode, + user, user_id, key_id: int, shell, ini_path: str, settings, env): + + # validate our settings for making a standalone calls + try: + self.service_api_host = settings['app.service_api.host'] + self.service_api_token = settings['app.service_api.token'] + except KeyError: + raise ImproperlyConfiguredError( + "app.service_api.host or app.service_api.token are missing. " + "Please ensure that app.service_api.host and app.service_api.token are " + "defined inside of .ini configuration file." + ) + + try: + self.api_url = settings['rhodecode.api.url'] + except KeyError: + raise ImproperlyConfiguredError( + "rhodecode.api.url is missing. " + "Please ensure that rhodecode.api.url is " + "defined inside of .ini configuration file." + ) + + super(SshWrapperStandalone, self).__init__( + command, connection_info, mode, user, user_id, key_id, shell, ini_path, settings, env) @staticmethod def parse_user_related_data(user_data): @@ -301,7 +326,7 @@ class SshWrapperStandalone(SshWrapper): exit_code = 1 elif scm_detected: - data = call_service_api(self.ini_path, { + data = call_service_api(self.service_api_host, self.service_api_token, self.api_url, { "method": "service_get_data_for_ssh_wrapper", "args": {"user_id": user_id, "repo_name": scm_repo, "key_id": self.key_id} }) @@ -339,7 +364,7 @@ class SshWrapperStandalone(SshWrapper): if repo_name.startswith('_'): org_repo_name = repo_name log.debug('translating UID repo %s', org_repo_name) - by_id_match = call_service_api(self.ini_path, { + by_id_match = call_service_api(self.service_api_host, self.service_api_token, self.api_url, { 'method': 'service_get_repo_name_by_id', "args": {"repo_id": repo_name} }) @@ -375,17 +400,17 @@ class SshWrapperStandalone(SshWrapper): server = MercurialServer( store=store, ini_path=self.ini_path, repo_name=repo, user=user, - user_permissions=permissions, config=self.config, env=self.env) + user_permissions=permissions, settings=self.settings, env=self.env) case 'git': server = GitServer( store=store, ini_path=self.ini_path, repo_name=repo, repo_mode=mode, user=user, - user_permissions=permissions, config=self.config, env=self.env) + user_permissions=permissions, settings=self.settings, env=self.env) case 'svn': server = SubversionServer( store=store, ini_path=self.ini_path, repo_name=None, user=user, - user_permissions=permissions, config=self.config, env=self.env) + user_permissions=permissions, settings=self.settings, env=self.env) case _: raise Exception(f'Unrecognised VCS: {vcs}') self.server_impl = server diff --git a/rhodecode/apps/ssh_support/lib/backends/base.py b/rhodecode/apps/ssh_support/lib/backends/base.py --- a/rhodecode/apps/ssh_support/lib/backends/base.py +++ b/rhodecode/apps/ssh_support/lib/backends/base.py @@ -20,27 +20,27 @@ import os import sys import logging -from rhodecode.lib.hooks_daemon import prepare_callback_daemon +from rhodecode.lib.hook_daemon.base import prepare_callback_daemon from rhodecode.lib.ext_json import sjson as json from rhodecode.lib.vcs.conf import settings as vcs_settings -from rhodecode.lib.utils import call_service_api -from rhodecode.model.scm import ScmModel +from rhodecode.lib.api_utils import call_service_api log = logging.getLogger(__name__) -class VcsServer(object): +class SSHVcsServer(object): repo_user_agent = None # set in child classes _path = None # set executable path for hg/git/svn binary backend = None # set in child classes tunnel = None # subprocess handling tunnel + settings = None # parsed settings module write_perms = ['repository.admin', 'repository.write'] read_perms = ['repository.read', 'repository.admin', 'repository.write'] - def __init__(self, user, user_permissions, config, env): + def __init__(self, user, user_permissions, settings, env): self.user = user self.user_permissions = user_permissions - self.config = config + self.settings = settings self.env = env self.stdin = sys.stdin @@ -59,9 +59,14 @@ class VcsServer(object): # Todo: Leave only "celery" case after transition. match self.hooks_protocol: case 'http': + from rhodecode.model.scm import ScmModel ScmModel().mark_for_invalidation(repo_name) case 'celery': - call_service_api(self.ini_path, { + service_api_host = self.settings['app.service_api.host'] + service_api_token = self.settings['app.service_api.token'] + api_url = self.settings['rhodecode.api.url'] + + call_service_api(service_api_host, service_api_token, api_url, { "method": "service_mark_for_invalidation", "args": {"repo_name": repo_name} }) @@ -118,7 +123,7 @@ class VcsServer(object): 'server_url': None, 'user_agent': f'{self.repo_user_agent}/ssh-user-agent', 'hooks': ['push', 'pull'], - 'hooks_module': 'rhodecode.lib.hooks_daemon', + 'hooks_module': 'rhodecode.lib.hook_daemon.hook_module', 'is_shadow_repo': False, 'detect_force_push': False, 'check_branch_perms': False, @@ -156,7 +161,7 @@ class VcsServer(object): return exit_code, action == "push" def run(self, tunnel_extras=None): - self.hooks_protocol = self.config.get('app:main', 'vcs.hooks.protocol') + self.hooks_protocol = self.settings['vcs.hooks.protocol'] tunnel_extras = tunnel_extras or {} extras = {} extras.update(tunnel_extras) diff --git a/rhodecode/apps/ssh_support/lib/backends/git.py b/rhodecode/apps/ssh_support/lib/backends/git.py --- a/rhodecode/apps/ssh_support/lib/backends/git.py +++ b/rhodecode/apps/ssh_support/lib/backends/git.py @@ -21,7 +21,7 @@ import logging import subprocess from vcsserver import hooks -from .base import VcsServer +from .base import SSHVcsServer log = logging.getLogger(__name__) @@ -70,19 +70,17 @@ class GitTunnelWrapper(object): return result -class GitServer(VcsServer): +class GitServer(SSHVcsServer): backend = 'git' repo_user_agent = 'git' - def __init__(self, store, ini_path, repo_name, repo_mode, - user, user_permissions, config, env): - super().\ - __init__(user, user_permissions, config, env) + def __init__(self, store, ini_path, repo_name, repo_mode, user, user_permissions, settings, env): + super().__init__(user, user_permissions, settings, env) self.store = store self.ini_path = ini_path self.repo_name = repo_name - self._path = self.git_path = config.get('app:main', 'ssh.executable.git') + self._path = self.git_path = settings['ssh.executable.git'] self.repo_mode = repo_mode self.tunnel = GitTunnelWrapper(server=self) diff --git a/rhodecode/apps/ssh_support/lib/backends/hg.py b/rhodecode/apps/ssh_support/lib/backends/hg.py --- a/rhodecode/apps/ssh_support/lib/backends/hg.py +++ b/rhodecode/apps/ssh_support/lib/backends/hg.py @@ -22,10 +22,10 @@ import logging import tempfile import textwrap import collections -from .base import VcsServer -from rhodecode.lib.utils import call_service_api -from rhodecode.model.db import RhodeCodeUi -from rhodecode.model.settings import VcsSettingsModel + +from .base import SSHVcsServer + +from rhodecode.lib.api_utils import call_service_api log = logging.getLogger(__name__) @@ -57,7 +57,7 @@ class MercurialTunnelWrapper(object): # cleanup custom hgrc file if os.path.isfile(hgrc_custom): with open(hgrc_custom, 'wb') as f: - f.write('') + f.write(b'') log.debug('Cleanup custom hgrc file under %s', hgrc_custom) # write temp @@ -94,62 +94,67 @@ class MercurialTunnelWrapper(object): self.remove_configs() -class MercurialServer(VcsServer): +class MercurialServer(SSHVcsServer): backend = 'hg' repo_user_agent = 'mercurial' cli_flags = ['phases', 'largefiles', 'extensions', 'experimental', 'hooks'] - def __init__(self, store, ini_path, repo_name, user, user_permissions, config, env): - super().__init__(user, user_permissions, config, env) + def __init__(self, store, ini_path, repo_name, user, user_permissions, settings, env): + super().__init__(user, user_permissions, settings, env) self.store = store self.ini_path = ini_path self.repo_name = repo_name - self._path = self.hg_path = config.get('app:main', 'ssh.executable.hg') + self._path = self.hg_path = settings['ssh.executable.hg'] self.tunnel = MercurialTunnelWrapper(server=self) def config_to_hgrc(self, repo_name): # Todo: once transition is done only call to service api should exist if self.hooks_protocol == 'celery': - data = call_service_api(self.ini_path, { + service_api_host = self.settings['app.service_api.host'] + service_api_token = self.settings['app.service_api.token'] + api_url = self.settings['rhodecode.api.url'] + data = call_service_api(service_api_host, service_api_token, api_url, { "method": "service_config_to_hgrc", "args": {"cli_flags": self.cli_flags, "repo_name": repo_name} }) return data['flags'] - - ui_sections = collections.defaultdict(list) - ui = VcsSettingsModel(repo=repo_name).get_ui_settings(section=None, key=None) + else: + from rhodecode.model.db import RhodeCodeUi + from rhodecode.model.settings import VcsSettingsModel + ui_sections = collections.defaultdict(list) + ui = VcsSettingsModel(repo=repo_name).get_ui_settings(section=None, key=None) - # write default hooks - default_hooks = [ - ('pretxnchangegroup.ssh_auth', 'python:vcsserver.hooks.pre_push_ssh_auth'), - ('pretxnchangegroup.ssh', 'python:vcsserver.hooks.pre_push_ssh'), - ('changegroup.ssh', 'python:vcsserver.hooks.post_push_ssh'), + # write default hooks + default_hooks = [ + ('pretxnchangegroup.ssh_auth', 'python:vcsserver.hooks.pre_push_ssh_auth'), + ('pretxnchangegroup.ssh', 'python:vcsserver.hooks.pre_push_ssh'), + ('changegroup.ssh', 'python:vcsserver.hooks.post_push_ssh'), - ('preoutgoing.ssh', 'python:vcsserver.hooks.pre_pull_ssh'), - ('outgoing.ssh', 'python:vcsserver.hooks.post_pull_ssh'), - ] + ('preoutgoing.ssh', 'python:vcsserver.hooks.pre_pull_ssh'), + ('outgoing.ssh', 'python:vcsserver.hooks.post_pull_ssh'), + ] - for k, v in default_hooks: - ui_sections['hooks'].append((k, v)) + for k, v in default_hooks: + ui_sections['hooks'].append((k, v)) - for entry in ui: - if not entry.active: - continue - sec = entry.section - key = entry.key - - if sec in self.cli_flags: - # we want only custom hooks, so we skip builtins - if sec == 'hooks' and key in RhodeCodeUi.HOOKS_BUILTIN: + for entry in ui: + if not entry.active: continue + sec = entry.section + key = entry.key - ui_sections[sec].append([key, entry.value]) + if sec in self.cli_flags: + # we want only custom hooks, so we skip builtins + if sec == 'hooks' and key in RhodeCodeUi.HOOKS_BUILTIN: + continue - flags = [] - for _sec, key_val in ui_sections.items(): - flags.append(' ') - flags.append(f'[{_sec}]') - for key, val in key_val: - flags.append(f'{key}= {val}') - return flags + ui_sections[sec].append([key, entry.value]) + + flags = [] + for _sec, key_val in ui_sections.items(): + flags.append(' ') + flags.append(f'[{_sec}]') + for key, val in key_val: + flags.append(f'{key}= {val}') + return flags diff --git a/rhodecode/apps/ssh_support/lib/backends/svn.py b/rhodecode/apps/ssh_support/lib/backends/svn.py --- a/rhodecode/apps/ssh_support/lib/backends/svn.py +++ b/rhodecode/apps/ssh_support/lib/backends/svn.py @@ -25,7 +25,7 @@ import tempfile from subprocess import Popen, PIPE import urllib.parse -from .base import VcsServer +from .base import SSHVcsServer log = logging.getLogger(__name__) @@ -218,20 +218,18 @@ class SubversionTunnelWrapper(object): return self.return_code -class SubversionServer(VcsServer): +class SubversionServer(SSHVcsServer): backend = 'svn' repo_user_agent = 'svn' - def __init__(self, store, ini_path, repo_name, - user, user_permissions, config, env): - super()\ - .__init__(user, user_permissions, config, env) + def __init__(self, store, ini_path, repo_name, user, user_permissions, settings, env): + super().__init__(user, user_permissions, settings, env) self.store = store self.ini_path = ini_path # NOTE(dan): repo_name at this point is empty, # this is set later in .run() based from parsed input stream self.repo_name = repo_name - self._path = self.svn_path = config.get('app:main', 'ssh.executable.svn') + self._path = self.svn_path = settings['ssh.executable.svn'] self.tunnel = SubversionTunnelWrapper(server=self) diff --git a/rhodecode/apps/ssh_support/lib/ssh_wrapper_v1.py b/rhodecode/apps/ssh_support/lib/ssh_wrapper_v1.py --- a/rhodecode/apps/ssh_support/lib/ssh_wrapper_v1.py +++ b/rhodecode/apps/ssh_support/lib/ssh_wrapper_v1.py @@ -31,7 +31,6 @@ from .utils import setup_custom_logging log = logging.getLogger(__name__) - @click.command() @click.argument('ini_path', type=click.Path(exists=True)) @click.option( @@ -55,11 +54,12 @@ def main(ini_path, mode, user, user_id, connection_info = os.environ.get('SSH_CONNECTION', '') time_start = time.time() with bootstrap(ini_path, env={'RC_CMD_SSH_WRAPPER': '1'}) as env: + settings = env['registry'].settings statsd = StatsdClient.statsd try: ssh_wrapper = SshWrapper( command, connection_info, mode, - user, user_id, key_id, shell, ini_path, env) + user, user_id, key_id, shell, ini_path, settings, env) except Exception: log.exception('Failed to execute SshWrapper') sys.exit(-5) diff --git a/rhodecode/apps/ssh_support/lib/ssh_wrapper_v2.py b/rhodecode/apps/ssh_support/lib/ssh_wrapper_v2.py --- a/rhodecode/apps/ssh_support/lib/ssh_wrapper_v2.py +++ b/rhodecode/apps/ssh_support/lib/ssh_wrapper_v2.py @@ -16,6 +16,14 @@ # RhodeCode Enterprise Edition, including its added features, Support services, # and proprietary license terms, please see https://rhodecode.com/licenses/ +""" +WARNING: be really carefully with changing ANY imports in this file +# This script is to mean as really fast executable, doing some imports here that would yield an import chain change +# can affect execution times... +# This can be easily debugged using such command:: +# time PYTHONPROFILEIMPORTTIME=1 rc-ssh-wrapper-v2 --debug --mode=test .dev/dev.ini +""" + import os import sys import time @@ -23,9 +31,12 @@ import logging import click +from rhodecode.config.config_maker import sanitize_settings_and_apply_defaults from rhodecode.lib.statsd_client import StatsdClient +from rhodecode.lib.config_utils import get_app_config_lightweight + +from .utils import setup_custom_logging from .backends import SshWrapperStandalone -from .utils import setup_custom_logging log = logging.getLogger(__name__) @@ -42,6 +53,8 @@ log = logging.getLogger(__name__) @click.option('--shell', '-s', is_flag=True, help='Allow Shell') @click.option('--debug', is_flag=True, help='Enabled detailed output logging') def main(ini_path, mode, user, user_id, key_id, shell, debug): + + time_start = time.time() setup_custom_logging(ini_path, debug) command = os.environ.get('SSH_ORIGINAL_COMMAND', '') @@ -50,21 +63,30 @@ def main(ini_path, mode, user, user_id, 'Unable to fetch SSH_ORIGINAL_COMMAND from environment.' 'Please make sure this is set and available during execution ' 'of this script.') - connection_info = os.environ.get('SSH_CONNECTION', '') - time_start = time.time() - env = {'RC_CMD_SSH_WRAPPER': '1'} + + # initialize settings and get defaults + settings = get_app_config_lightweight(ini_path) + settings = sanitize_settings_and_apply_defaults({'__file__': ini_path}, settings) + + # init and bootstrap StatsdClient + StatsdClient.setup(settings) statsd = StatsdClient.statsd + try: + connection_info = os.environ.get('SSH_CONNECTION', '') + env = {'RC_CMD_SSH_WRAPPER': '1'} ssh_wrapper = SshWrapperStandalone( command, connection_info, mode, - user, user_id, key_id, shell, ini_path, env) + user, user_id, key_id, shell, ini_path, settings, env) except Exception: log.exception('Failed to execute SshWrapper') sys.exit(-5) + return_code = ssh_wrapper.wrap() operation_took = time.time() - time_start if statsd: operation_took_ms = round(1000.0 * operation_took) statsd.timing("rhodecode_ssh_wrapper_timing.histogram", operation_took_ms, use_decimals=False) + sys.exit(return_code) diff --git a/rhodecode/apps/ssh_support/lib/utils.py b/rhodecode/apps/ssh_support/lib/utils.py --- a/rhodecode/apps/ssh_support/lib/utils.py +++ b/rhodecode/apps/ssh_support/lib/utils.py @@ -17,11 +17,11 @@ # and proprietary license terms, please see https://rhodecode.com/licenses/ import logging -from pyramid.paster import setup_logging def setup_custom_logging(ini_path, debug): if debug: + from pyramid.paster import setup_logging # Lazy import # enabled rhodecode.ini controlled logging setup setup_logging(ini_path) else: diff --git a/rhodecode/apps/ssh_support/tests/conftest.py b/rhodecode/apps/ssh_support/tests/conftest.py --- a/rhodecode/apps/ssh_support/tests/conftest.py +++ b/rhodecode/apps/ssh_support/tests/conftest.py @@ -52,7 +52,10 @@ def dummy_env(): def plain_dummy_user(): - return AttributeDict(username='test_user') + return AttributeDict( + user_id=1, + username='test_user' + ) @pytest.fixture() @@ -65,4 +68,4 @@ def ssh_wrapper(app, dummy_conf_file, du conn_info = '127.0.0.1 22 10.0.0.1 443' return SshWrapper( 'random command', conn_info, 'auto', 'admin', '1', key_id='1', - shell=False, ini_path=dummy_conf_file, env=dummy_env) + shell=False, ini_path=dummy_conf_file, settings={}, env=dummy_env) diff --git a/rhodecode/apps/ssh_support/tests/test_server_git.py b/rhodecode/apps/ssh_support/tests/test_server_git.py --- a/rhodecode/apps/ssh_support/tests/test_server_git.py +++ b/rhodecode/apps/ssh_support/tests/test_server_git.py @@ -25,6 +25,7 @@ from rhodecode.apps.ssh_support.lib.back from rhodecode.apps.ssh_support.tests.conftest import plain_dummy_env, plain_dummy_user from rhodecode.lib.ext_json import json + class GitServerCreator(object): root = '/tmp/repo/path/' git_path = '/usr/local/bin/git' @@ -39,10 +40,7 @@ class GitServerCreator(object): user = plain_dummy_user() def __init__(self): - def config_get(part, key): - return self.config_data.get(part, {}).get(key) - self.config_mock = mock.Mock() - self.config_mock.get = mock.Mock(side_effect=config_get) + pass def create(self, **kwargs): parameters = { @@ -54,7 +52,7 @@ class GitServerCreator(object): 'user_permissions': { self.repo_name: 'repository.admin' }, - 'config': self.config_mock, + 'settings': self.config_data['app:main'], 'env': plain_dummy_env() } parameters.update(kwargs) @@ -142,7 +140,7 @@ class TestGitServer(object): 'server_url': None, 'hooks': ['push', 'pull'], 'is_shadow_repo': False, - 'hooks_module': 'rhodecode.lib.hooks_daemon', + 'hooks_module': 'rhodecode.lib.hook_daemon.hook_module', 'check_branch_perms': False, 'detect_force_push': False, 'user_agent': u'git/ssh-user-agent', diff --git a/rhodecode/apps/ssh_support/tests/test_server_hg.py b/rhodecode/apps/ssh_support/tests/test_server_hg.py --- a/rhodecode/apps/ssh_support/tests/test_server_hg.py +++ b/rhodecode/apps/ssh_support/tests/test_server_hg.py @@ -38,10 +38,7 @@ class MercurialServerCreator(object): user = plain_dummy_user() def __init__(self): - def config_get(part, key): - return self.config_data.get(part, {}).get(key) - self.config_mock = mock.Mock() - self.config_mock.get = mock.Mock(side_effect=config_get) + pass def create(self, **kwargs): parameters = { @@ -52,7 +49,7 @@ class MercurialServerCreator(object): 'user_permissions': { 'test_hg': 'repository.admin' }, - 'config': self.config_mock, + 'settings': self.config_data['app:main'], 'env': plain_dummy_env() } parameters.update(kwargs) diff --git a/rhodecode/apps/ssh_support/tests/test_server_svn.py b/rhodecode/apps/ssh_support/tests/test_server_svn.py --- a/rhodecode/apps/ssh_support/tests/test_server_svn.py +++ b/rhodecode/apps/ssh_support/tests/test_server_svn.py @@ -36,10 +36,7 @@ class SubversionServerCreator(object): user = plain_dummy_user() def __init__(self): - def config_get(part, key): - return self.config_data.get(part, {}).get(key) - self.config_mock = mock.Mock() - self.config_mock.get = mock.Mock(side_effect=config_get) + pass def create(self, **kwargs): parameters = { @@ -50,7 +47,7 @@ class SubversionServerCreator(object): 'user_permissions': { self.repo_name: 'repository.admin' }, - 'config': self.config_mock, + 'settings': self.config_data['app:main'], 'env': plain_dummy_env() } diff --git a/rhodecode/apps/ssh_support/tests/test_ssh_wrapper.py b/rhodecode/apps/ssh_support/tests/test_ssh_wrapper.py --- a/rhodecode/apps/ssh_support/tests/test_ssh_wrapper.py +++ b/rhodecode/apps/ssh_support/tests/test_ssh_wrapper.py @@ -28,10 +28,6 @@ class TestSSHWrapper(object): permissions={}, branch_permissions={}) assert str(exc_info.value) == 'Unrecognised VCS: microsoft-tfs' - def test_parse_config(self, ssh_wrapper): - config = ssh_wrapper.parse_config(ssh_wrapper.ini_path) - assert config - def test_get_connection_info(self, ssh_wrapper): conn_info = ssh_wrapper.get_connection_info() assert {'client_ip': '127.0.0.1', diff --git a/rhodecode/config/config_maker.py b/rhodecode/config/config_maker.py new file mode 100644 --- /dev/null +++ b/rhodecode/config/config_maker.py @@ -0,0 +1,198 @@ +# 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 . +# +# 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 tempfile +import logging + +from pyramid.settings import asbool + +from rhodecode.config.settings_maker import SettingsMaker +from rhodecode.config import utils as config_utils + +log = logging.getLogger(__name__) + + +def sanitize_settings_and_apply_defaults(global_config, settings): + """ + Applies settings defaults and does all type conversion. + + We would move all settings parsing and preparation into this place, so that + we have only one place left which deals with this part. The remaining parts + of the application would start to rely fully on well-prepared settings. + + This piece would later be split up per topic to avoid a big fat monster + function. + """ + jn = os.path.join + + global_settings_maker = SettingsMaker(global_config) + global_settings_maker.make_setting('debug', default=False, parser='bool') + debug_enabled = asbool(global_config.get('debug')) + + settings_maker = SettingsMaker(settings) + + settings_maker.make_setting( + 'logging.autoconfigure', + default=False, + parser='bool') + + logging_conf = jn(os.path.dirname(global_config.get('__file__')), 'logging.ini') + settings_maker.enable_logging(logging_conf, level='INFO' if debug_enabled else 'DEBUG') + + # Default includes, possible to change as a user + pyramid_includes = settings_maker.make_setting('pyramid.includes', [], parser='list:newline') + log.debug( + "Using the following pyramid.includes: %s", + pyramid_includes) + + settings_maker.make_setting('rhodecode.edition', 'Community Edition') + settings_maker.make_setting('rhodecode.edition_id', 'CE') + + if 'mako.default_filters' not in settings: + # set custom default filters if we don't have it defined + settings['mako.imports'] = 'from rhodecode.lib.base import h_filter' + settings['mako.default_filters'] = 'h_filter' + + if 'mako.directories' not in settings: + mako_directories = settings.setdefault('mako.directories', [ + # Base templates of the original application + 'rhodecode:templates', + ]) + log.debug( + "Using the following Mako template directories: %s", + mako_directories) + + # NOTE(marcink): fix redis requirement for schema of connection since 3.X + if 'beaker.session.type' in settings and settings['beaker.session.type'] == 'ext:redis': + raw_url = settings['beaker.session.url'] + if not raw_url.startswith(('redis://', 'rediss://', 'unix://')): + settings['beaker.session.url'] = 'redis://' + raw_url + + settings_maker.make_setting('__file__', global_config.get('__file__')) + + # TODO: johbo: Re-think this, usually the call to config.include + # should allow to pass in a prefix. + settings_maker.make_setting('rhodecode.api.url', '/_admin/api') + + # Sanitize generic settings. + settings_maker.make_setting('default_encoding', 'UTF-8', parser='list') + settings_maker.make_setting('is_test', False, parser='bool') + settings_maker.make_setting('gzip_responses', False, parser='bool') + + # statsd + settings_maker.make_setting('statsd.enabled', False, parser='bool') + settings_maker.make_setting('statsd.statsd_host', 'statsd-exporter', parser='string') + settings_maker.make_setting('statsd.statsd_port', 9125, parser='int') + settings_maker.make_setting('statsd.statsd_prefix', '') + settings_maker.make_setting('statsd.statsd_ipv6', False, parser='bool') + + settings_maker.make_setting('vcs.svn.compatible_version', '') + settings_maker.make_setting('vcs.hooks.protocol', 'http') + settings_maker.make_setting('vcs.hooks.host', '*') + settings_maker.make_setting('vcs.scm_app_implementation', 'http') + settings_maker.make_setting('vcs.server', '') + settings_maker.make_setting('vcs.server.protocol', 'http') + settings_maker.make_setting('vcs.server.enable', 'true', parser='bool') + settings_maker.make_setting('startup.import_repos', 'false', parser='bool') + settings_maker.make_setting('vcs.hooks.direct_calls', 'false', parser='bool') + settings_maker.make_setting('vcs.start_server', 'false', parser='bool') + settings_maker.make_setting('vcs.backends', 'hg, git, svn', parser='list') + settings_maker.make_setting('vcs.connection_timeout', 3600, parser='int') + + settings_maker.make_setting('vcs.methods.cache', True, parser='bool') + + # Support legacy values of vcs.scm_app_implementation. Legacy + # configurations may use 'rhodecode.lib.middleware.utils.scm_app_http', or + # disabled since 4.13 'vcsserver.scm_app' which is now mapped to 'http'. + scm_app_impl = settings['vcs.scm_app_implementation'] + if scm_app_impl in ['rhodecode.lib.middleware.utils.scm_app_http', 'vcsserver.scm_app']: + settings['vcs.scm_app_implementation'] = 'http' + + settings_maker.make_setting('appenlight', False, parser='bool') + + temp_store = tempfile.gettempdir() + tmp_cache_dir = jn(temp_store, 'rc_cache') + + # save default, cache dir, and use it for all backends later. + default_cache_dir = settings_maker.make_setting( + 'cache_dir', + default=tmp_cache_dir, default_when_empty=True, + parser='dir:ensured') + + # exception store cache + settings_maker.make_setting( + 'exception_tracker.store_path', + default=jn(default_cache_dir, 'exc_store'), default_when_empty=True, + parser='dir:ensured' + ) + + settings_maker.make_setting( + 'celerybeat-schedule.path', + default=jn(default_cache_dir, 'celerybeat_schedule', 'celerybeat-schedule.db'), default_when_empty=True, + parser='file:ensured' + ) + + settings_maker.make_setting('exception_tracker.send_email', False, parser='bool') + settings_maker.make_setting('exception_tracker.email_prefix', '[RHODECODE ERROR]', default_when_empty=True) + + # sessions, ensure file since no-value is memory + settings_maker.make_setting('beaker.session.type', 'file') + settings_maker.make_setting('beaker.session.data_dir', jn(default_cache_dir, 'session_data')) + + # cache_general + settings_maker.make_setting('rc_cache.cache_general.backend', 'dogpile.cache.rc.file_namespace') + settings_maker.make_setting('rc_cache.cache_general.expiration_time', 60 * 60 * 12, parser='int') + settings_maker.make_setting('rc_cache.cache_general.arguments.filename', jn(default_cache_dir, 'rhodecode_cache_general.db')) + + # cache_perms + settings_maker.make_setting('rc_cache.cache_perms.backend', 'dogpile.cache.rc.file_namespace') + settings_maker.make_setting('rc_cache.cache_perms.expiration_time', 60 * 60, parser='int') + settings_maker.make_setting('rc_cache.cache_perms.arguments.filename', jn(default_cache_dir, 'rhodecode_cache_perms_db')) + + # cache_repo + settings_maker.make_setting('rc_cache.cache_repo.backend', 'dogpile.cache.rc.file_namespace') + settings_maker.make_setting('rc_cache.cache_repo.expiration_time', 60 * 60 * 24 * 30, parser='int') + settings_maker.make_setting('rc_cache.cache_repo.arguments.filename', jn(default_cache_dir, 'rhodecode_cache_repo_db')) + + # cache_license + settings_maker.make_setting('rc_cache.cache_license.backend', 'dogpile.cache.rc.file_namespace') + settings_maker.make_setting('rc_cache.cache_license.expiration_time', 60 * 5, parser='int') + settings_maker.make_setting('rc_cache.cache_license.arguments.filename', jn(default_cache_dir, 'rhodecode_cache_license_db')) + + # cache_repo_longterm memory, 96H + settings_maker.make_setting('rc_cache.cache_repo_longterm.backend', 'dogpile.cache.rc.memory_lru') + settings_maker.make_setting('rc_cache.cache_repo_longterm.expiration_time', 345600, parser='int') + settings_maker.make_setting('rc_cache.cache_repo_longterm.max_size', 10000, parser='int') + + # sql_cache_short + settings_maker.make_setting('rc_cache.sql_cache_short.backend', 'dogpile.cache.rc.memory_lru') + settings_maker.make_setting('rc_cache.sql_cache_short.expiration_time', 30, parser='int') + settings_maker.make_setting('rc_cache.sql_cache_short.max_size', 10000, parser='int') + + # archive_cache + settings_maker.make_setting('archive_cache.store_dir', jn(default_cache_dir, 'archive_cache'), default_when_empty=True,) + settings_maker.make_setting('archive_cache.cache_size_gb', 10, parser='float') + settings_maker.make_setting('archive_cache.cache_shards', 10, parser='int') + + settings_maker.env_expand() + + # configure instance id + config_utils.set_instance_id(settings) + + return settings diff --git a/rhodecode/config/middleware.py b/rhodecode/config/middleware.py --- a/rhodecode/config/middleware.py +++ b/rhodecode/config/middleware.py @@ -19,7 +19,7 @@ import os import sys import collections -import tempfile + import time import logging.config @@ -32,14 +32,13 @@ from pyramid.httpexceptions import ( HTTPException, HTTPError, HTTPInternalServerError, HTTPFound, HTTPNotFound) from pyramid.renderers import render_to_response -from rhodecode import api from rhodecode.model import meta from rhodecode.config import patches -from rhodecode.config import utils as config_utils -from rhodecode.config.settings_maker import SettingsMaker + from rhodecode.config.environment import load_pyramid_environment import rhodecode.events +from rhodecode.config.config_maker import sanitize_settings_and_apply_defaults from rhodecode.lib.middleware.vcs import VCSMiddleware from rhodecode.lib.request import Request from rhodecode.lib.vcs import VCSCommunicationError @@ -465,173 +464,3 @@ def wrap_app_in_wsgi_middlewares(pyramid log.debug('Request processing finalized: %.4fs', total) return pyramid_app_with_cleanup - - -def sanitize_settings_and_apply_defaults(global_config, settings): - """ - Applies settings defaults and does all type conversion. - - We would move all settings parsing and preparation into this place, so that - we have only one place left which deals with this part. The remaining parts - of the application would start to rely fully on well prepared settings. - - This piece would later be split up per topic to avoid a big fat monster - function. - """ - jn = os.path.join - - global_settings_maker = SettingsMaker(global_config) - global_settings_maker.make_setting('debug', default=False, parser='bool') - debug_enabled = asbool(global_config.get('debug')) - - settings_maker = SettingsMaker(settings) - - settings_maker.make_setting( - 'logging.autoconfigure', - default=False, - parser='bool') - - logging_conf = jn(os.path.dirname(global_config.get('__file__')), 'logging.ini') - settings_maker.enable_logging(logging_conf, level='INFO' if debug_enabled else 'DEBUG') - - # Default includes, possible to change as a user - pyramid_includes = settings_maker.make_setting('pyramid.includes', [], parser='list:newline') - log.debug( - "Using the following pyramid.includes: %s", - pyramid_includes) - - settings_maker.make_setting('rhodecode.edition', 'Community Edition') - settings_maker.make_setting('rhodecode.edition_id', 'CE') - - if 'mako.default_filters' not in settings: - # set custom default filters if we don't have it defined - settings['mako.imports'] = 'from rhodecode.lib.base import h_filter' - settings['mako.default_filters'] = 'h_filter' - - if 'mako.directories' not in settings: - mako_directories = settings.setdefault('mako.directories', [ - # Base templates of the original application - 'rhodecode:templates', - ]) - log.debug( - "Using the following Mako template directories: %s", - mako_directories) - - # NOTE(marcink): fix redis requirement for schema of connection since 3.X - if 'beaker.session.type' in settings and settings['beaker.session.type'] == 'ext:redis': - raw_url = settings['beaker.session.url'] - if not raw_url.startswith(('redis://', 'rediss://', 'unix://')): - settings['beaker.session.url'] = 'redis://' + raw_url - - settings_maker.make_setting('__file__', global_config.get('__file__')) - - # TODO: johbo: Re-think this, usually the call to config.include - # should allow to pass in a prefix. - settings_maker.make_setting('rhodecode.api.url', api.DEFAULT_URL) - - # Sanitize generic settings. - settings_maker.make_setting('default_encoding', 'UTF-8', parser='list') - settings_maker.make_setting('is_test', False, parser='bool') - settings_maker.make_setting('gzip_responses', False, parser='bool') - - # statsd - settings_maker.make_setting('statsd.enabled', False, parser='bool') - settings_maker.make_setting('statsd.statsd_host', 'statsd-exporter', parser='string') - settings_maker.make_setting('statsd.statsd_port', 9125, parser='int') - settings_maker.make_setting('statsd.statsd_prefix', '') - settings_maker.make_setting('statsd.statsd_ipv6', False, parser='bool') - - settings_maker.make_setting('vcs.svn.compatible_version', '') - settings_maker.make_setting('vcs.hooks.protocol', 'http') - settings_maker.make_setting('vcs.hooks.host', '*') - settings_maker.make_setting('vcs.scm_app_implementation', 'http') - settings_maker.make_setting('vcs.server', '') - settings_maker.make_setting('vcs.server.protocol', 'http') - settings_maker.make_setting('vcs.server.enable', 'true', parser='bool') - settings_maker.make_setting('startup.import_repos', 'false', parser='bool') - settings_maker.make_setting('vcs.hooks.direct_calls', 'false', parser='bool') - settings_maker.make_setting('vcs.start_server', 'false', parser='bool') - settings_maker.make_setting('vcs.backends', 'hg, git, svn', parser='list') - settings_maker.make_setting('vcs.connection_timeout', 3600, parser='int') - - settings_maker.make_setting('vcs.methods.cache', True, parser='bool') - - # Support legacy values of vcs.scm_app_implementation. Legacy - # configurations may use 'rhodecode.lib.middleware.utils.scm_app_http', or - # disabled since 4.13 'vcsserver.scm_app' which is now mapped to 'http'. - scm_app_impl = settings['vcs.scm_app_implementation'] - if scm_app_impl in ['rhodecode.lib.middleware.utils.scm_app_http', 'vcsserver.scm_app']: - settings['vcs.scm_app_implementation'] = 'http' - - settings_maker.make_setting('appenlight', False, parser='bool') - - temp_store = tempfile.gettempdir() - tmp_cache_dir = jn(temp_store, 'rc_cache') - - # save default, cache dir, and use it for all backends later. - default_cache_dir = settings_maker.make_setting( - 'cache_dir', - default=tmp_cache_dir, default_when_empty=True, - parser='dir:ensured') - - # exception store cache - settings_maker.make_setting( - 'exception_tracker.store_path', - default=jn(default_cache_dir, 'exc_store'), default_when_empty=True, - parser='dir:ensured' - ) - - settings_maker.make_setting( - 'celerybeat-schedule.path', - default=jn(default_cache_dir, 'celerybeat_schedule', 'celerybeat-schedule.db'), default_when_empty=True, - parser='file:ensured' - ) - - settings_maker.make_setting('exception_tracker.send_email', False, parser='bool') - settings_maker.make_setting('exception_tracker.email_prefix', '[RHODECODE ERROR]', default_when_empty=True) - - # sessions, ensure file since no-value is memory - settings_maker.make_setting('beaker.session.type', 'file') - settings_maker.make_setting('beaker.session.data_dir', jn(default_cache_dir, 'session_data')) - - # cache_general - settings_maker.make_setting('rc_cache.cache_general.backend', 'dogpile.cache.rc.file_namespace') - settings_maker.make_setting('rc_cache.cache_general.expiration_time', 60 * 60 * 12, parser='int') - settings_maker.make_setting('rc_cache.cache_general.arguments.filename', jn(default_cache_dir, 'rhodecode_cache_general.db')) - - # cache_perms - settings_maker.make_setting('rc_cache.cache_perms.backend', 'dogpile.cache.rc.file_namespace') - settings_maker.make_setting('rc_cache.cache_perms.expiration_time', 60 * 60, parser='int') - settings_maker.make_setting('rc_cache.cache_perms.arguments.filename', jn(default_cache_dir, 'rhodecode_cache_perms_db')) - - # cache_repo - settings_maker.make_setting('rc_cache.cache_repo.backend', 'dogpile.cache.rc.file_namespace') - settings_maker.make_setting('rc_cache.cache_repo.expiration_time', 60 * 60 * 24 * 30, parser='int') - settings_maker.make_setting('rc_cache.cache_repo.arguments.filename', jn(default_cache_dir, 'rhodecode_cache_repo_db')) - - # cache_license - settings_maker.make_setting('rc_cache.cache_license.backend', 'dogpile.cache.rc.file_namespace') - settings_maker.make_setting('rc_cache.cache_license.expiration_time', 60 * 5, parser='int') - settings_maker.make_setting('rc_cache.cache_license.arguments.filename', jn(default_cache_dir, 'rhodecode_cache_license_db')) - - # cache_repo_longterm memory, 96H - settings_maker.make_setting('rc_cache.cache_repo_longterm.backend', 'dogpile.cache.rc.memory_lru') - settings_maker.make_setting('rc_cache.cache_repo_longterm.expiration_time', 345600, parser='int') - settings_maker.make_setting('rc_cache.cache_repo_longterm.max_size', 10000, parser='int') - - # sql_cache_short - settings_maker.make_setting('rc_cache.sql_cache_short.backend', 'dogpile.cache.rc.memory_lru') - settings_maker.make_setting('rc_cache.sql_cache_short.expiration_time', 30, parser='int') - settings_maker.make_setting('rc_cache.sql_cache_short.max_size', 10000, parser='int') - - # archive_cache - settings_maker.make_setting('archive_cache.store_dir', jn(default_cache_dir, 'archive_cache'), default_when_empty=True,) - settings_maker.make_setting('archive_cache.cache_size_gb', 10, parser='float') - settings_maker.make_setting('archive_cache.cache_shards', 10, parser='int') - - settings_maker.env_expand() - - # configure instance id - config_utils.set_instance_id(settings) - - return settings diff --git a/rhodecode/config/utils.py b/rhodecode/config/utils.py --- a/rhodecode/config/utils.py +++ b/rhodecode/config/utils.py @@ -19,8 +19,6 @@ import os import platform -from rhodecode.model import init_model - def configure_vcs(config): """ @@ -44,6 +42,7 @@ def configure_vcs(config): 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)) diff --git a/rhodecode/lib/api_utils.py b/rhodecode/lib/api_utils.py new file mode 100644 --- /dev/null +++ b/rhodecode/lib/api_utils.py @@ -0,0 +1,38 @@ +# 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 . +# +# 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 urllib.parse + +from rhodecode.lib.vcs import CurlSession +from rhodecode.lib.ext_json import json + + +def call_service_api(service_api_host, service_api_token, api_url, payload): + + payload.update({ + 'id': 'service', + 'auth_token': service_api_token + }) + + service_api_url = urllib.parse.urljoin(service_api_host, api_url) + response = CurlSession().post(service_api_url, json.dumps(payload)) + + if response.status_code != 200: + raise Exception(f"Service API at {service_api_url} responded with error: {response.status_code}") + + return json.loads(response.content)['result'] diff --git a/rhodecode/lib/base.py b/rhodecode/lib/base.py --- a/rhodecode/lib/base.py +++ b/rhodecode/lib/base.py @@ -567,7 +567,7 @@ def add_events_routes(config): def bootstrap_config(request, registry_name='RcTestRegistry'): - from rhodecode.config.middleware import sanitize_settings_and_apply_defaults + from rhodecode.config.config_maker import sanitize_settings_and_apply_defaults import pyramid.testing registry = pyramid.testing.Registry(registry_name) diff --git a/rhodecode/lib/config_utils.py b/rhodecode/lib/config_utils.py new file mode 100644 --- /dev/null +++ b/rhodecode/lib/config_utils.py @@ -0,0 +1,40 @@ +# 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 . +# +# 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 + + +def get_config(ini_path, **kwargs): + import configparser + parser = configparser.ConfigParser(**kwargs) + parser.read(ini_path) + return parser + + +def get_app_config_lightweight(ini_path): + parser = get_config(ini_path) + parser.set('app:main', 'here', os.getcwd()) + parser.set('app:main', '__file__', ini_path) + return dict(parser.items('app:main')) + + +def get_app_config(ini_path): + """ + This loads the app context and provides a heavy type iniliaziation of config + """ + from paste.deploy.loadwsgi import appconfig + return appconfig(f'config:{ini_path}', relative_to=os.getcwd()) diff --git a/rhodecode/lib/hook_daemon/__init__.py b/rhodecode/lib/hook_daemon/__init__.py new file mode 100644 --- /dev/null +++ b/rhodecode/lib/hook_daemon/__init__.py @@ -0,0 +1,17 @@ +# 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 . +# +# 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/ diff --git a/rhodecode/lib/hook_daemon/base.py b/rhodecode/lib/hook_daemon/base.py new file mode 100644 --- /dev/null +++ b/rhodecode/lib/hook_daemon/base.py @@ -0,0 +1,115 @@ +# 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 . +# +# 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 time +import logging +import tempfile + +from rhodecode.lib.config_utils import get_config +from rhodecode.lib.ext_json import json + +log = logging.getLogger(__name__) + + +class BaseHooksCallbackDaemon: + """ + Basic context manager for actions that don't require some extra + """ + def __init__(self): + pass + + def __enter__(self): + log.debug('Running `%s` callback daemon', self.__class__.__name__) + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + log.debug('Exiting `%s` callback daemon', self.__class__.__name__) + + +class HooksModuleCallbackDaemon(BaseHooksCallbackDaemon): + + def __init__(self, module): + super().__init__() + self.hooks_module = module + + +def get_txn_id_data_path(txn_id): + import rhodecode + + root = rhodecode.CONFIG.get('cache_dir') or tempfile.gettempdir() + final_dir = os.path.join(root, 'svn_txn_id') + + if not os.path.isdir(final_dir): + os.makedirs(final_dir) + return os.path.join(final_dir, 'rc_txn_id_{}'.format(txn_id)) + + +def store_txn_id_data(txn_id, data_dict): + if not txn_id: + log.warning('Cannot store txn_id because it is empty') + return + + path = get_txn_id_data_path(txn_id) + try: + with open(path, 'wb') as f: + f.write(json.dumps(data_dict)) + except Exception: + log.exception('Failed to write txn_id metadata') + + +def get_txn_id_from_store(txn_id): + """ + Reads txn_id from store and if present returns the data for callback manager + """ + path = get_txn_id_data_path(txn_id) + try: + with open(path, 'rb') as f: + return json.loads(f.read()) + except Exception: + return {} + + +def prepare_callback_daemon(extras, protocol, host, txn_id=None): + txn_details = get_txn_id_from_store(txn_id) + port = txn_details.get('port', 0) + match protocol: + case 'http': + from rhodecode.lib.hook_daemon.http_hooks_deamon import HttpHooksCallbackDaemon + callback_daemon = HttpHooksCallbackDaemon( + txn_id=txn_id, host=host, port=port) + case 'celery': + from rhodecode.lib.hook_daemon.celery_hooks_deamon import CeleryHooksCallbackDaemon + callback_daemon = CeleryHooksCallbackDaemon(get_config(extras['config'])) + case 'local': + from rhodecode.lib.hook_daemon.hook_module import Hooks + callback_daemon = HooksModuleCallbackDaemon(Hooks.__module__) + case _: + log.error('Unsupported callback daemon protocol "%s"', protocol) + raise Exception('Unsupported callback daemon protocol.') + + extras['hooks_uri'] = getattr(callback_daemon, 'hooks_uri', '') + extras['task_queue'] = getattr(callback_daemon, 'task_queue', '') + extras['task_backend'] = getattr(callback_daemon, 'task_backend', '') + extras['hooks_protocol'] = protocol + extras['time'] = time.time() + + # register txn_id + extras['txn_id'] = txn_id + log.debug('Prepared a callback daemon: %s', + callback_daemon.__class__.__name__) + return callback_daemon, extras diff --git a/rhodecode/lib/hook_daemon/celery_hooks_deamon.py b/rhodecode/lib/hook_daemon/celery_hooks_deamon.py new file mode 100644 --- /dev/null +++ b/rhodecode/lib/hook_daemon/celery_hooks_deamon.py @@ -0,0 +1,30 @@ +# 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 . +# +# 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/ + +from rhodecode.lib.hook_daemon.base import BaseHooksCallbackDaemon + + +class CeleryHooksCallbackDaemon(BaseHooksCallbackDaemon): + """ + Context manger for achieving a compatibility with celery backend + """ + + def __init__(self, config): + # TODO: replace this with settings bootstrapped... + self.task_queue = config.get('app:main', 'celery.broker_url') + self.task_backend = config.get('app:main', 'celery.result_backend') diff --git a/rhodecode/lib/hook_daemon/hook_module.py b/rhodecode/lib/hook_daemon/hook_module.py new file mode 100644 --- /dev/null +++ b/rhodecode/lib/hook_daemon/hook_module.py @@ -0,0 +1,104 @@ +# 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 . +# +# 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 logging +import traceback + +from rhodecode.model import meta + +from rhodecode.lib import hooks_base +from rhodecode.lib.exceptions import HTTPLockedRC, HTTPBranchProtected +from rhodecode.lib.utils2 import AttributeDict + +log = logging.getLogger(__name__) + + +class Hooks(object): + """ + Exposes the hooks for remote callbacks + """ + def __init__(self, request=None, log_prefix=''): + self.log_prefix = log_prefix + self.request = request + + def repo_size(self, extras): + log.debug("%sCalled repo_size of %s object", self.log_prefix, self) + return self._call_hook(hooks_base.repo_size, extras) + + def pre_pull(self, extras): + log.debug("%sCalled pre_pull of %s object", self.log_prefix, self) + return self._call_hook(hooks_base.pre_pull, extras) + + def post_pull(self, extras): + log.debug("%sCalled post_pull of %s object", self.log_prefix, self) + return self._call_hook(hooks_base.post_pull, extras) + + def pre_push(self, extras): + log.debug("%sCalled pre_push of %s object", self.log_prefix, self) + return self._call_hook(hooks_base.pre_push, extras) + + def post_push(self, extras): + log.debug("%sCalled post_push of %s object", self.log_prefix, self) + return self._call_hook(hooks_base.post_push, extras) + + def _call_hook(self, hook, extras): + extras = AttributeDict(extras) + _server_url = extras['server_url'] + + extras.request = self.request + + try: + result = hook(extras) + if result is None: + raise Exception(f'Failed to obtain hook result from func: {hook}') + except HTTPBranchProtected as handled_error: + # Those special cases don't need error reporting. It's a case of + # locked repo or protected branch + result = AttributeDict({ + 'status': handled_error.code, + 'output': handled_error.explanation + }) + except (HTTPLockedRC, Exception) as error: + # locked needs different handling since we need to also + # handle PULL operations + exc_tb = '' + if not isinstance(error, HTTPLockedRC): + exc_tb = traceback.format_exc() + log.exception('%sException when handling hook %s', self.log_prefix, hook) + error_args = error.args + return { + 'status': 128, + 'output': '', + 'exception': type(error).__name__, + 'exception_traceback': exc_tb, + 'exception_args': error_args, + } + finally: + meta.Session.remove() + + log.debug('%sGot hook call response %s', self.log_prefix, result) + return { + 'status': result.status, + 'output': result.output, + } + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + pass diff --git a/rhodecode/lib/hook_daemon/http_hooks_deamon.py b/rhodecode/lib/hook_daemon/http_hooks_deamon.py new file mode 100644 --- /dev/null +++ b/rhodecode/lib/hook_daemon/http_hooks_deamon.py @@ -0,0 +1,280 @@ +# 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 . +# +# 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 logging +import traceback +import threading +import socket +import msgpack +import gevent + +from http.server import BaseHTTPRequestHandler +from socketserver import TCPServer + +from rhodecode.model import meta +from rhodecode.lib.ext_json import json +from rhodecode.lib import rc_cache +from rhodecode.lib.hook_daemon.base import get_txn_id_data_path +from rhodecode.lib.hook_daemon.hook_module import Hooks + +log = logging.getLogger(__name__) + + +class HooksHttpHandler(BaseHTTPRequestHandler): + + JSON_HOOKS_PROTO = 'json.v1' + MSGPACK_HOOKS_PROTO = 'msgpack.v1' + # starting with RhodeCode 5.0.0 MsgPack is the default, prior it used json + DEFAULT_HOOKS_PROTO = MSGPACK_HOOKS_PROTO + + @classmethod + def serialize_data(cls, data, proto=DEFAULT_HOOKS_PROTO): + if proto == cls.MSGPACK_HOOKS_PROTO: + return msgpack.packb(data) + return json.dumps(data) + + @classmethod + def deserialize_data(cls, data, proto=DEFAULT_HOOKS_PROTO): + if proto == cls.MSGPACK_HOOKS_PROTO: + return msgpack.unpackb(data) + return json.loads(data) + + def do_POST(self): + hooks_proto, method, extras = self._read_request() + log.debug('Handling HooksHttpHandler %s with %s proto', method, hooks_proto) + + txn_id = getattr(self.server, 'txn_id', None) + if txn_id: + log.debug('Computing TXN_ID based on `%s`:`%s`', + extras['repository'], extras['txn_id']) + computed_txn_id = rc_cache.utils.compute_key_from_params( + extras['repository'], extras['txn_id']) + if txn_id != computed_txn_id: + raise Exception( + 'TXN ID fail: expected {} got {} instead'.format( + txn_id, computed_txn_id)) + + request = getattr(self.server, 'request', None) + try: + hooks = Hooks(request=request, log_prefix='HOOKS: {} '.format(self.server.server_address)) + result = self._call_hook_method(hooks, method, extras) + + except Exception as e: + exc_tb = traceback.format_exc() + result = { + 'exception': e.__class__.__name__, + 'exception_traceback': exc_tb, + 'exception_args': e.args + } + self._write_response(hooks_proto, result) + + def _read_request(self): + length = int(self.headers['Content-Length']) + # respect sent headers, fallback to OLD proto for compatability + hooks_proto = self.headers.get('rc-hooks-protocol') or self.JSON_HOOKS_PROTO + if hooks_proto == self.MSGPACK_HOOKS_PROTO: + # support for new vcsserver msgpack based protocol hooks + body = self.rfile.read(length) + data = self.deserialize_data(body) + else: + body = self.rfile.read(length) + data = self.deserialize_data(body) + + return hooks_proto, data['method'], data['extras'] + + def _write_response(self, hooks_proto, result): + self.send_response(200) + if hooks_proto == self.MSGPACK_HOOKS_PROTO: + self.send_header("Content-type", "application/msgpack") + self.end_headers() + data = self.serialize_data(result) + self.wfile.write(data) + else: + self.send_header("Content-type", "text/json") + self.end_headers() + data = self.serialize_data(result) + self.wfile.write(data) + + def _call_hook_method(self, hooks, method, extras): + try: + result = getattr(hooks, method)(extras) + finally: + meta.Session.remove() + return result + + def log_message(self, format, *args): + """ + This is an overridden method of BaseHTTPRequestHandler which logs using + a logging library instead of writing directly to stderr. + """ + + message = format % args + + log.debug( + "HOOKS: client=%s - - [%s] %s", self.client_address, + self.log_date_time_string(), message) + + +class ThreadedHookCallbackDaemon(object): + + _callback_thread = None + _daemon = None + _done = False + use_gevent = False + + def __init__(self, txn_id=None, host=None, port=None): + self._prepare(txn_id=txn_id, host=host, port=port) + if self.use_gevent: + self._run_func = self._run_gevent + self._stop_func = self._stop_gevent + else: + self._run_func = self._run + self._stop_func = self._stop + + def __enter__(self): + log.debug('Running `%s` callback daemon', self.__class__.__name__) + self._run_func() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + log.debug('Exiting `%s` callback daemon', self.__class__.__name__) + self._stop_func() + + def _prepare(self, txn_id=None, host=None, port=None): + raise NotImplementedError() + + def _run(self): + raise NotImplementedError() + + def _stop(self): + raise NotImplementedError() + + def _run_gevent(self): + raise NotImplementedError() + + def _stop_gevent(self): + raise NotImplementedError() + + +class HttpHooksCallbackDaemon(ThreadedHookCallbackDaemon): + """ + Context manager which will run a callback daemon in a background thread. + """ + + hooks_uri = None + + # From Python docs: Polling reduces our responsiveness to a shutdown + # request and wastes cpu at all other times. + POLL_INTERVAL = 0.01 + + use_gevent = False + + @property + def _hook_prefix(self): + return 'HOOKS: {} '.format(self.hooks_uri) + + def get_hostname(self): + return socket.gethostname() or '127.0.0.1' + + def get_available_port(self, min_port=20000, max_port=65535): + from rhodecode.lib.utils2 import get_available_port as _get_port + return _get_port(min_port, max_port) + + def _prepare(self, txn_id=None, host=None, port=None): + from pyramid.threadlocal import get_current_request + + if not host or host == "*": + host = self.get_hostname() + if not port: + port = self.get_available_port() + + server_address = (host, port) + self.hooks_uri = '{}:{}'.format(host, port) + self.txn_id = txn_id + self._done = False + + log.debug( + "%s Preparing HTTP callback daemon registering hook object: %s", + self._hook_prefix, HooksHttpHandler) + + self._daemon = TCPServer(server_address, HooksHttpHandler) + # inject transaction_id for later verification + self._daemon.txn_id = self.txn_id + + # pass the WEB app request into daemon + self._daemon.request = get_current_request() + + def _run(self): + log.debug("Running thread-based loop of callback daemon in background") + callback_thread = threading.Thread( + target=self._daemon.serve_forever, + kwargs={'poll_interval': self.POLL_INTERVAL}) + callback_thread.daemon = True + callback_thread.start() + self._callback_thread = callback_thread + + def _run_gevent(self): + log.debug("Running gevent-based loop of callback daemon in background") + # create a new greenlet for the daemon's serve_forever method + callback_greenlet = gevent.spawn( + self._daemon.serve_forever, + poll_interval=self.POLL_INTERVAL) + + # store reference to greenlet + self._callback_greenlet = callback_greenlet + + # switch to this greenlet + gevent.sleep(0.01) + + def _stop(self): + log.debug("Waiting for background thread to finish.") + self._daemon.shutdown() + self._callback_thread.join() + self._daemon = None + self._callback_thread = None + if self.txn_id: + txn_id_file = get_txn_id_data_path(self.txn_id) + log.debug('Cleaning up TXN ID %s', txn_id_file) + if os.path.isfile(txn_id_file): + os.remove(txn_id_file) + + log.debug("Background thread done.") + + def _stop_gevent(self): + log.debug("Waiting for background greenlet to finish.") + + # if greenlet exists and is running + if self._callback_greenlet and not self._callback_greenlet.dead: + # shutdown daemon if it exists + if self._daemon: + self._daemon.shutdown() + + # kill the greenlet + self._callback_greenlet.kill() + + self._daemon = None + self._callback_greenlet = None + + if self.txn_id: + txn_id_file = get_txn_id_data_path(self.txn_id) + log.debug('Cleaning up TXN ID %s', txn_id_file) + if os.path.isfile(txn_id_file): + os.remove(txn_id_file) + + log.debug("Background greenlet done.") diff --git a/rhodecode/lib/hooks_daemon.py b/rhodecode/lib/hooks_daemon.py deleted file mode 100644 --- a/rhodecode/lib/hooks_daemon.py +++ /dev/null @@ -1,451 +0,0 @@ -# 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 . -# -# 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 time -import logging -import tempfile -import traceback -import threading -import socket -import msgpack -import gevent - -from http.server import BaseHTTPRequestHandler -from socketserver import TCPServer - -import rhodecode -from rhodecode.lib.exceptions import HTTPLockedRC, HTTPBranchProtected -from rhodecode.model import meta -from rhodecode.lib import hooks_base -from rhodecode.lib.utils2 import AttributeDict -from rhodecode.lib.pyramid_utils import get_config -from rhodecode.lib.ext_json import json -from rhodecode.lib import rc_cache - -log = logging.getLogger(__name__) - - -class HooksHttpHandler(BaseHTTPRequestHandler): - - JSON_HOOKS_PROTO = 'json.v1' - MSGPACK_HOOKS_PROTO = 'msgpack.v1' - # starting with RhodeCode 5.0.0 MsgPack is the default, prior it used json - DEFAULT_HOOKS_PROTO = MSGPACK_HOOKS_PROTO - - @classmethod - def serialize_data(cls, data, proto=DEFAULT_HOOKS_PROTO): - if proto == cls.MSGPACK_HOOKS_PROTO: - return msgpack.packb(data) - return json.dumps(data) - - @classmethod - def deserialize_data(cls, data, proto=DEFAULT_HOOKS_PROTO): - if proto == cls.MSGPACK_HOOKS_PROTO: - return msgpack.unpackb(data) - return json.loads(data) - - def do_POST(self): - hooks_proto, method, extras = self._read_request() - log.debug('Handling HooksHttpHandler %s with %s proto', method, hooks_proto) - - txn_id = getattr(self.server, 'txn_id', None) - if txn_id: - log.debug('Computing TXN_ID based on `%s`:`%s`', - extras['repository'], extras['txn_id']) - computed_txn_id = rc_cache.utils.compute_key_from_params( - extras['repository'], extras['txn_id']) - if txn_id != computed_txn_id: - raise Exception( - 'TXN ID fail: expected {} got {} instead'.format( - txn_id, computed_txn_id)) - - request = getattr(self.server, 'request', None) - try: - hooks = Hooks(request=request, log_prefix='HOOKS: {} '.format(self.server.server_address)) - result = self._call_hook_method(hooks, method, extras) - - except Exception as e: - exc_tb = traceback.format_exc() - result = { - 'exception': e.__class__.__name__, - 'exception_traceback': exc_tb, - 'exception_args': e.args - } - self._write_response(hooks_proto, result) - - def _read_request(self): - length = int(self.headers['Content-Length']) - # respect sent headers, fallback to OLD proto for compatability - hooks_proto = self.headers.get('rc-hooks-protocol') or self.JSON_HOOKS_PROTO - if hooks_proto == self.MSGPACK_HOOKS_PROTO: - # support for new vcsserver msgpack based protocol hooks - body = self.rfile.read(length) - data = self.deserialize_data(body) - else: - body = self.rfile.read(length) - data = self.deserialize_data(body) - - return hooks_proto, data['method'], data['extras'] - - def _write_response(self, hooks_proto, result): - self.send_response(200) - if hooks_proto == self.MSGPACK_HOOKS_PROTO: - self.send_header("Content-type", "application/msgpack") - self.end_headers() - data = self.serialize_data(result) - self.wfile.write(data) - else: - self.send_header("Content-type", "text/json") - self.end_headers() - data = self.serialize_data(result) - self.wfile.write(data) - - def _call_hook_method(self, hooks, method, extras): - try: - result = getattr(hooks, method)(extras) - finally: - meta.Session.remove() - return result - - def log_message(self, format, *args): - """ - This is an overridden method of BaseHTTPRequestHandler which logs using - logging library instead of writing directly to stderr. - """ - - message = format % args - - log.debug( - "HOOKS: client=%s - - [%s] %s", self.client_address, - self.log_date_time_string(), message) - - -class BaseHooksCallbackDaemon: - """ - Basic context manager for actions that don't require some extra - """ - def __init__(self): - self.hooks_module = Hooks.__module__ - - def __enter__(self): - log.debug('Running `%s` callback daemon', self.__class__.__name__) - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - log.debug('Exiting `%s` callback daemon', self.__class__.__name__) - - -class CeleryHooksCallbackDaemon(BaseHooksCallbackDaemon): - """ - Context manger for achieving a compatibility with celery backend - """ - - def __init__(self, config): - self.task_queue = config.get('app:main', 'celery.broker_url') - self.task_backend = config.get('app:main', 'celery.result_backend') - - -class ThreadedHookCallbackDaemon(object): - - _callback_thread = None - _daemon = None - _done = False - use_gevent = False - - def __init__(self, txn_id=None, host=None, port=None): - self._prepare(txn_id=txn_id, host=host, port=port) - if self.use_gevent: - self._run_func = self._run_gevent - self._stop_func = self._stop_gevent - else: - self._run_func = self._run - self._stop_func = self._stop - - def __enter__(self): - log.debug('Running `%s` callback daemon', self.__class__.__name__) - self._run_func() - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - log.debug('Exiting `%s` callback daemon', self.__class__.__name__) - self._stop_func() - - def _prepare(self, txn_id=None, host=None, port=None): - raise NotImplementedError() - - def _run(self): - raise NotImplementedError() - - def _stop(self): - raise NotImplementedError() - - def _run_gevent(self): - raise NotImplementedError() - - def _stop_gevent(self): - raise NotImplementedError() - - -class HttpHooksCallbackDaemon(ThreadedHookCallbackDaemon): - """ - Context manager which will run a callback daemon in a background thread. - """ - - hooks_uri = None - - # From Python docs: Polling reduces our responsiveness to a shutdown - # request and wastes cpu at all other times. - POLL_INTERVAL = 0.01 - - use_gevent = False - - @property - def _hook_prefix(self): - return 'HOOKS: {} '.format(self.hooks_uri) - - def get_hostname(self): - return socket.gethostname() or '127.0.0.1' - - def get_available_port(self, min_port=20000, max_port=65535): - from rhodecode.lib.utils2 import get_available_port as _get_port - return _get_port(min_port, max_port) - - def _prepare(self, txn_id=None, host=None, port=None): - from pyramid.threadlocal import get_current_request - - if not host or host == "*": - host = self.get_hostname() - if not port: - port = self.get_available_port() - - server_address = (host, port) - self.hooks_uri = '{}:{}'.format(host, port) - self.txn_id = txn_id - self._done = False - - log.debug( - "%s Preparing HTTP callback daemon registering hook object: %s", - self._hook_prefix, HooksHttpHandler) - - self._daemon = TCPServer(server_address, HooksHttpHandler) - # inject transaction_id for later verification - self._daemon.txn_id = self.txn_id - - # pass the WEB app request into daemon - self._daemon.request = get_current_request() - - def _run(self): - log.debug("Running thread-based loop of callback daemon in background") - callback_thread = threading.Thread( - target=self._daemon.serve_forever, - kwargs={'poll_interval': self.POLL_INTERVAL}) - callback_thread.daemon = True - callback_thread.start() - self._callback_thread = callback_thread - - def _run_gevent(self): - log.debug("Running gevent-based loop of callback daemon in background") - # create a new greenlet for the daemon's serve_forever method - callback_greenlet = gevent.spawn( - self._daemon.serve_forever, - poll_interval=self.POLL_INTERVAL) - - # store reference to greenlet - self._callback_greenlet = callback_greenlet - - # switch to this greenlet - gevent.sleep(0.01) - - def _stop(self): - log.debug("Waiting for background thread to finish.") - self._daemon.shutdown() - self._callback_thread.join() - self._daemon = None - self._callback_thread = None - if self.txn_id: - txn_id_file = get_txn_id_data_path(self.txn_id) - log.debug('Cleaning up TXN ID %s', txn_id_file) - if os.path.isfile(txn_id_file): - os.remove(txn_id_file) - - log.debug("Background thread done.") - - def _stop_gevent(self): - log.debug("Waiting for background greenlet to finish.") - - # if greenlet exists and is running - if self._callback_greenlet and not self._callback_greenlet.dead: - # shutdown daemon if it exists - if self._daemon: - self._daemon.shutdown() - - # kill the greenlet - self._callback_greenlet.kill() - - self._daemon = None - self._callback_greenlet = None - - if self.txn_id: - txn_id_file = get_txn_id_data_path(self.txn_id) - log.debug('Cleaning up TXN ID %s', txn_id_file) - if os.path.isfile(txn_id_file): - os.remove(txn_id_file) - - log.debug("Background greenlet done.") - - -def get_txn_id_data_path(txn_id): - import rhodecode - - root = rhodecode.CONFIG.get('cache_dir') or tempfile.gettempdir() - final_dir = os.path.join(root, 'svn_txn_id') - - if not os.path.isdir(final_dir): - os.makedirs(final_dir) - return os.path.join(final_dir, 'rc_txn_id_{}'.format(txn_id)) - - -def store_txn_id_data(txn_id, data_dict): - if not txn_id: - log.warning('Cannot store txn_id because it is empty') - return - - path = get_txn_id_data_path(txn_id) - try: - with open(path, 'wb') as f: - f.write(json.dumps(data_dict)) - except Exception: - log.exception('Failed to write txn_id metadata') - - -def get_txn_id_from_store(txn_id): - """ - Reads txn_id from store and if present returns the data for callback manager - """ - path = get_txn_id_data_path(txn_id) - try: - with open(path, 'rb') as f: - return json.loads(f.read()) - except Exception: - return {} - - -def prepare_callback_daemon(extras, protocol, host, txn_id=None): - txn_details = get_txn_id_from_store(txn_id) - port = txn_details.get('port', 0) - match protocol: - case 'http': - callback_daemon = HttpHooksCallbackDaemon( - txn_id=txn_id, host=host, port=port) - case 'celery': - callback_daemon = CeleryHooksCallbackDaemon(get_config(extras['config'])) - case 'local': - callback_daemon = BaseHooksCallbackDaemon() - case _: - log.error('Unsupported callback daemon protocol "%s"', protocol) - raise Exception('Unsupported callback daemon protocol.') - - extras['hooks_uri'] = getattr(callback_daemon, 'hooks_uri', '') - extras['task_queue'] = getattr(callback_daemon, 'task_queue', '') - extras['task_backend'] = getattr(callback_daemon, 'task_backend', '') - extras['hooks_protocol'] = protocol - extras['time'] = time.time() - - # register txn_id - extras['txn_id'] = txn_id - log.debug('Prepared a callback daemon: %s', - callback_daemon.__class__.__name__) - return callback_daemon, extras - - -class Hooks(object): - """ - Exposes the hooks for remote call backs - """ - def __init__(self, request=None, log_prefix=''): - self.log_prefix = log_prefix - self.request = request - - def repo_size(self, extras): - log.debug("%sCalled repo_size of %s object", self.log_prefix, self) - return self._call_hook(hooks_base.repo_size, extras) - - def pre_pull(self, extras): - log.debug("%sCalled pre_pull of %s object", self.log_prefix, self) - return self._call_hook(hooks_base.pre_pull, extras) - - def post_pull(self, extras): - log.debug("%sCalled post_pull of %s object", self.log_prefix, self) - return self._call_hook(hooks_base.post_pull, extras) - - def pre_push(self, extras): - log.debug("%sCalled pre_push of %s object", self.log_prefix, self) - return self._call_hook(hooks_base.pre_push, extras) - - def post_push(self, extras): - log.debug("%sCalled post_push of %s object", self.log_prefix, self) - return self._call_hook(hooks_base.post_push, extras) - - def _call_hook(self, hook, extras): - extras = AttributeDict(extras) - server_url = extras['server_url'] - - extras.request = self.request - - try: - result = hook(extras) - if result is None: - raise Exception( - 'Failed to obtain hook result from func: {}'.format(hook)) - except HTTPBranchProtected as handled_error: - # Those special cases doesn't need error reporting. It's a case of - # locked repo or protected branch - result = AttributeDict({ - 'status': handled_error.code, - 'output': handled_error.explanation - }) - except (HTTPLockedRC, Exception) as error: - # locked needs different handling since we need to also - # handle PULL operations - exc_tb = '' - if not isinstance(error, HTTPLockedRC): - exc_tb = traceback.format_exc() - log.exception('%sException when handling hook %s', self.log_prefix, hook) - error_args = error.args - return { - 'status': 128, - 'output': '', - 'exception': type(error).__name__, - 'exception_traceback': exc_tb, - 'exception_args': error_args, - } - finally: - meta.Session.remove() - - log.debug('%sGot hook call response %s', self.log_prefix, result) - return { - 'status': result.status, - 'output': result.output, - } - - def __enter__(self): - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - pass diff --git a/rhodecode/lib/middleware/simplesvn.py b/rhodecode/lib/middleware/simplesvn.py --- a/rhodecode/lib/middleware/simplesvn.py +++ b/rhodecode/lib/middleware/simplesvn.py @@ -34,8 +34,7 @@ from rhodecode.lib.utils import is_valid from rhodecode.lib.str_utils import safe_str, safe_int, safe_bytes from rhodecode.lib.type_utils import str2bool from rhodecode.lib.ext_json import json -from rhodecode.lib.hooks_daemon import store_txn_id_data - +from rhodecode.lib.hook_daemon.base import store_txn_id_data log = logging.getLogger(__name__) diff --git a/rhodecode/lib/middleware/simplevcs.py b/rhodecode/lib/middleware/simplevcs.py --- a/rhodecode/lib/middleware/simplevcs.py +++ b/rhodecode/lib/middleware/simplevcs.py @@ -45,7 +45,7 @@ from rhodecode.lib.auth import AuthUser, from rhodecode.lib.base import ( BasicAuth, get_ip_addr, get_user_agent, vcs_operation_context) from rhodecode.lib.exceptions import (UserCreationError, NotAllowedToCreateUserError) -from rhodecode.lib.hooks_daemon import prepare_callback_daemon +from rhodecode.lib.hook_daemon.base import prepare_callback_daemon from rhodecode.lib.middleware import appenlight from rhodecode.lib.middleware.utils import scm_app_http from rhodecode.lib.str_utils import safe_bytes diff --git a/rhodecode/lib/pyramid_utils.py b/rhodecode/lib/pyramid_utils.py --- a/rhodecode/lib/pyramid_utils.py +++ b/rhodecode/lib/pyramid_utils.py @@ -20,24 +20,14 @@ import os import configparser + +from rhodecode.lib.config_utils import get_config from pyramid.paster import bootstrap as pyramid_bootstrap, setup_logging # pragma: no cover -from rhodecode.lib.request import Request - - -def get_config(ini_path, **kwargs): - parser = configparser.ConfigParser(**kwargs) - parser.read(ini_path) - return parser - - -def get_app_config(ini_path): - from paste.deploy.loadwsgi import appconfig - return appconfig(f'config:{ini_path}', relative_to=os.getcwd()) - def bootstrap(config_uri, options=None, env=None): from rhodecode.lib.utils2 import AttributeDict + from rhodecode.lib.request import Request if env: os.environ.update(env) diff --git a/rhodecode/lib/rc_commands/setup_rc.py b/rhodecode/lib/rc_commands/setup_rc.py --- a/rhodecode/lib/rc_commands/setup_rc.py +++ b/rhodecode/lib/rc_commands/setup_rc.py @@ -20,7 +20,8 @@ import logging import click import pyramid.paster -from rhodecode.lib.pyramid_utils import bootstrap, get_app_config +from rhodecode.lib.pyramid_utils import bootstrap +from rhodecode.lib.config_utils import get_app_config from rhodecode.lib.db_manage import DbManage from rhodecode.lib.utils2 import get_encryption_key from rhodecode.model.db import Session diff --git a/rhodecode/lib/utils.py b/rhodecode/lib/utils.py --- a/rhodecode/lib/utils.py +++ b/rhodecode/lib/utils.py @@ -32,11 +32,9 @@ import socket import tempfile import traceback import tarfile -import urllib.parse -import warnings + from functools import wraps from os.path import join as jn -from configparser import NoOptionError import paste import pkg_resources @@ -55,9 +53,6 @@ from rhodecode.model import meta from rhodecode.model.db import ( Repository, User, RhodeCodeUi, UserLog, RepoGroup, UserGroup) from rhodecode.model.meta import Session -from rhodecode.lib.pyramid_utils import get_config -from rhodecode.lib.vcs import CurlSession -from rhodecode.lib.vcs.exceptions import ImproperlyConfiguredError log = logging.getLogger(__name__) @@ -827,33 +822,3 @@ def send_test_email(recipients, email_bo email_body = email_body_plaintext = email_body subject = f'SUBJECT FROM: {socket.gethostname()}' tasks.send_email(recipients, subject, email_body_plaintext, email_body) - - -def call_service_api(ini_path, payload): - config = get_config(ini_path) - try: - host = config.get('app:main', 'app.service_api.host') - except NoOptionError: - raise ImproperlyConfiguredError( - "app.service_api.host is missing. " - "Please ensure that app.service_api.host and app.service_api.token are " - "defined inside of .ini configuration file." - ) - try: - api_url = config.get('app:main', 'rhodecode.api.url') - except NoOptionError: - from rhodecode import api - log.debug('Cannot find rhodecode.api.url, setting API URL TO Default value') - api_url = api.DEFAULT_URL - - payload.update({ - 'id': 'service', - 'auth_token': config.get('app:main', 'app.service_api.token') - }) - - response = CurlSession().post(urllib.parse.urljoin(host, api_url), json.dumps(payload)) - - if response.status_code != 200: - raise Exception("Service API responded with error") - - return json.loads(response.content)['result'] diff --git a/rhodecode/model/pull_request.py b/rhodecode/model/pull_request.py --- a/rhodecode/model/pull_request.py +++ b/rhodecode/model/pull_request.py @@ -38,7 +38,7 @@ from rhodecode.translation import lazy_u from rhodecode.lib import helpers as h, hooks_utils, diffs from rhodecode.lib import audit_logger from collections import OrderedDict -from rhodecode.lib.hooks_daemon import prepare_callback_daemon +from rhodecode.lib.hook_daemon.base import prepare_callback_daemon from rhodecode.lib.ext_json import sjson as json from rhodecode.lib.markup_renderer import ( DEFAULT_COMMENTS_RENDERER, RstTemplateRenderer) diff --git a/rhodecode/tests/fixture_mods/fixture_pyramid.py b/rhodecode/tests/fixture_mods/fixture_pyramid.py --- a/rhodecode/tests/fixture_mods/fixture_pyramid.py +++ b/rhodecode/tests/fixture_mods/fixture_pyramid.py @@ -19,7 +19,7 @@ import pytest -from rhodecode.lib.pyramid_utils import get_app_config +from rhodecode.lib.config_utils import get_app_config from rhodecode.tests.fixture import TestINI from rhodecode.tests.server_utils import RcVCSServer diff --git a/rhodecode/tests/fixture_mods/fixture_utils.py b/rhodecode/tests/fixture_mods/fixture_utils.py --- a/rhodecode/tests/fixture_mods/fixture_utils.py +++ b/rhodecode/tests/fixture_mods/fixture_utils.py @@ -174,7 +174,7 @@ def http_environ(): @pytest.fixture(scope='session') def baseapp(ini_config, vcsserver, http_environ_session): - from rhodecode.lib.pyramid_utils import get_app_config + from rhodecode.lib.config_utils import get_app_config from rhodecode.config.middleware import make_pyramid_app log.info("Using the RhodeCode configuration:{}".format(ini_config)) diff --git a/rhodecode/tests/lib/test_hooks_daemon.py b/rhodecode/tests/lib/test_hooks_daemon.py --- a/rhodecode/tests/lib/test_hooks_daemon.py +++ b/rhodecode/tests/lib/test_hooks_daemon.py @@ -25,17 +25,20 @@ import msgpack import pytest import tempfile -from rhodecode.lib import hooks_daemon +from rhodecode.lib.hook_daemon import http_hooks_deamon +from rhodecode.lib.hook_daemon import celery_hooks_deamon +from rhodecode.lib.hook_daemon import hook_module +from rhodecode.lib.hook_daemon import base as hook_base from rhodecode.lib.str_utils import safe_bytes from rhodecode.tests.utils import assert_message_in_log from rhodecode.lib.ext_json import json -test_proto = hooks_daemon.HooksHttpHandler.MSGPACK_HOOKS_PROTO +test_proto = http_hooks_deamon.HooksHttpHandler.MSGPACK_HOOKS_PROTO class TestHooks(object): def test_hooks_can_be_used_as_a_context_processor(self): - hooks = hooks_daemon.Hooks() + hooks = hook_module.Hooks() with hooks as return_value: pass assert hooks == return_value @@ -52,10 +55,10 @@ class TestHooksHttpHandler(object): } request = self._generate_post_request(data) hooks_patcher = mock.patch.object( - hooks_daemon.Hooks, data['method'], create=True, return_value=1) + hook_module.Hooks, data['method'], create=True, return_value=1) with hooks_patcher as hooks_mock: - handler = hooks_daemon.HooksHttpHandler + handler = http_hooks_deamon.HooksHttpHandler handler.DEFAULT_HOOKS_PROTO = test_proto handler.wbufsize = 10240 MockServer(handler, request) @@ -73,21 +76,21 @@ class TestHooksHttpHandler(object): # patching our _read to return test method and proto used read_patcher = mock.patch.object( - hooks_daemon.HooksHttpHandler, '_read_request', + http_hooks_deamon.HooksHttpHandler, '_read_request', return_value=(test_proto, rpc_method, extras)) # patch Hooks instance to return hook_result data on 'test' call hooks_patcher = mock.patch.object( - hooks_daemon.Hooks, rpc_method, create=True, + hook_module.Hooks, rpc_method, create=True, return_value=hook_result) with read_patcher, hooks_patcher: - handler = hooks_daemon.HooksHttpHandler + handler = http_hooks_deamon.HooksHttpHandler handler.DEFAULT_HOOKS_PROTO = test_proto handler.wbufsize = 10240 server = MockServer(handler, request) - expected_result = hooks_daemon.HooksHttpHandler.serialize_data(hook_result) + expected_result = http_hooks_deamon.HooksHttpHandler.serialize_data(hook_result) server.request.output_stream.seek(0) assert server.request.output_stream.readlines()[-1] == expected_result @@ -97,15 +100,15 @@ class TestHooksHttpHandler(object): rpc_method = 'test' read_patcher = mock.patch.object( - hooks_daemon.HooksHttpHandler, '_read_request', + http_hooks_deamon.HooksHttpHandler, '_read_request', return_value=(test_proto, rpc_method, {})) hooks_patcher = mock.patch.object( - hooks_daemon.Hooks, rpc_method, create=True, + hook_module.Hooks, rpc_method, create=True, side_effect=Exception('Test exception')) with read_patcher, hooks_patcher: - handler = hooks_daemon.HooksHttpHandler + handler = http_hooks_deamon.HooksHttpHandler handler.DEFAULT_HOOKS_PROTO = test_proto handler.wbufsize = 10240 server = MockServer(handler, request) @@ -113,7 +116,7 @@ class TestHooksHttpHandler(object): server.request.output_stream.seek(0) data = server.request.output_stream.readlines() msgpack_data = b''.join(data[5:]) - org_exc = hooks_daemon.HooksHttpHandler.deserialize_data(msgpack_data) + org_exc = http_hooks_deamon.HooksHttpHandler.deserialize_data(msgpack_data) expected_result = { 'exception': 'Exception', 'exception_traceback': org_exc['exception_traceback'], @@ -123,8 +126,7 @@ class TestHooksHttpHandler(object): def test_log_message_writes_to_debug_log(self, caplog): ip_port = ('0.0.0.0', 8888) - handler = hooks_daemon.HooksHttpHandler( - MockRequest('POST /'), ip_port, mock.Mock()) + handler = http_hooks_deamon.HooksHttpHandler(MockRequest('POST /'), ip_port, mock.Mock()) fake_date = '1/Nov/2015 00:00:00' date_patcher = mock.patch.object( handler, 'log_date_time_string', return_value=fake_date) @@ -136,10 +138,10 @@ class TestHooksHttpHandler(object): assert_message_in_log( caplog.records, expected_message, - levelno=logging.DEBUG, module='hooks_daemon') + levelno=logging.DEBUG, module='http_hooks_deamon') def _generate_post_request(self, data, proto=test_proto): - if proto == hooks_daemon.HooksHttpHandler.MSGPACK_HOOKS_PROTO: + if proto == http_hooks_deamon.HooksHttpHandler.MSGPACK_HOOKS_PROTO: payload = msgpack.packb(data) else: payload = json.dumps(data) @@ -151,18 +153,18 @@ class TestHooksHttpHandler(object): class ThreadedHookCallbackDaemon(object): def test_constructor_calls_prepare(self): prepare_daemon_patcher = mock.patch.object( - hooks_daemon.ThreadedHookCallbackDaemon, '_prepare') + http_hooks_deamon.ThreadedHookCallbackDaemon, '_prepare') with prepare_daemon_patcher as prepare_daemon_mock: - hooks_daemon.ThreadedHookCallbackDaemon() + http_hooks_deamon.ThreadedHookCallbackDaemon() prepare_daemon_mock.assert_called_once_with() def test_run_is_called_on_context_start(self): patchers = mock.patch.multiple( - hooks_daemon.ThreadedHookCallbackDaemon, + http_hooks_deamon.ThreadedHookCallbackDaemon, _run=mock.DEFAULT, _prepare=mock.DEFAULT, __exit__=mock.DEFAULT) with patchers as mocks: - daemon = hooks_daemon.ThreadedHookCallbackDaemon() + daemon = http_hooks_deamon.ThreadedHookCallbackDaemon() with daemon as daemon_context: pass mocks['_run'].assert_called_once_with() @@ -170,11 +172,11 @@ class ThreadedHookCallbackDaemon(object) def test_stop_is_called_on_context_exit(self): patchers = mock.patch.multiple( - hooks_daemon.ThreadedHookCallbackDaemon, + http_hooks_deamon.ThreadedHookCallbackDaemon, _run=mock.DEFAULT, _prepare=mock.DEFAULT, _stop=mock.DEFAULT) with patchers as mocks: - daemon = hooks_daemon.ThreadedHookCallbackDaemon() + daemon = http_hooks_deamon.ThreadedHookCallbackDaemon() with daemon as daemon_context: assert mocks['_stop'].call_count == 0 @@ -185,46 +187,47 @@ class ThreadedHookCallbackDaemon(object) class TestHttpHooksCallbackDaemon(object): def test_hooks_callback_generates_new_port(self, caplog): with caplog.at_level(logging.DEBUG): - daemon = hooks_daemon.HttpHooksCallbackDaemon(host='127.0.0.1', port=8881) + daemon = http_hooks_deamon.HttpHooksCallbackDaemon(host='127.0.0.1', port=8881) assert daemon._daemon.server_address == ('127.0.0.1', 8881) with caplog.at_level(logging.DEBUG): - daemon = hooks_daemon.HttpHooksCallbackDaemon(host=None, port=None) + daemon = http_hooks_deamon.HttpHooksCallbackDaemon(host=None, port=None) assert daemon._daemon.server_address[1] in range(0, 66000) assert daemon._daemon.server_address[0] != '127.0.0.1' def test_prepare_inits_daemon_variable(self, tcp_server, caplog): with self._tcp_patcher(tcp_server), caplog.at_level(logging.DEBUG): - daemon = hooks_daemon.HttpHooksCallbackDaemon(host='127.0.0.1', port=8881) + daemon = http_hooks_deamon.HttpHooksCallbackDaemon(host='127.0.0.1', port=8881) assert daemon._daemon == tcp_server _, port = tcp_server.server_address msg = f"HOOKS: 127.0.0.1:{port} Preparing HTTP callback daemon registering " \ - f"hook object: " + f"hook object: " assert_message_in_log( - caplog.records, msg, levelno=logging.DEBUG, module='hooks_daemon') + caplog.records, msg, levelno=logging.DEBUG, module='http_hooks_deamon') def test_prepare_inits_hooks_uri_and_logs_it( self, tcp_server, caplog): with self._tcp_patcher(tcp_server), caplog.at_level(logging.DEBUG): - daemon = hooks_daemon.HttpHooksCallbackDaemon(host='127.0.0.1', port=8881) + daemon = http_hooks_deamon.HttpHooksCallbackDaemon(host='127.0.0.1', port=8881) _, port = tcp_server.server_address expected_uri = '{}:{}'.format('127.0.0.1', port) assert daemon.hooks_uri == expected_uri msg = f"HOOKS: 127.0.0.1:{port} Preparing HTTP callback daemon registering " \ - f"hook object: " + f"hook object: " + assert_message_in_log( caplog.records, msg, - levelno=logging.DEBUG, module='hooks_daemon') + levelno=logging.DEBUG, module='http_hooks_deamon') def test_run_creates_a_thread(self, tcp_server): thread = mock.Mock() with self._tcp_patcher(tcp_server): - daemon = hooks_daemon.HttpHooksCallbackDaemon() + daemon = http_hooks_deamon.HttpHooksCallbackDaemon() with self._thread_patcher(thread) as thread_mock: daemon._run() @@ -238,7 +241,7 @@ class TestHttpHooksCallbackDaemon(object def test_run_logs(self, tcp_server, caplog): with self._tcp_patcher(tcp_server): - daemon = hooks_daemon.HttpHooksCallbackDaemon() + daemon = http_hooks_deamon.HttpHooksCallbackDaemon() with self._thread_patcher(mock.Mock()), caplog.at_level(logging.DEBUG): daemon._run() @@ -246,13 +249,13 @@ class TestHttpHooksCallbackDaemon(object assert_message_in_log( caplog.records, 'Running thread-based loop of callback daemon in background', - levelno=logging.DEBUG, module='hooks_daemon') + levelno=logging.DEBUG, module='http_hooks_deamon') def test_stop_cleans_up_the_connection(self, tcp_server, caplog): thread = mock.Mock() with self._tcp_patcher(tcp_server): - daemon = hooks_daemon.HttpHooksCallbackDaemon() + daemon = http_hooks_deamon.HttpHooksCallbackDaemon() with self._thread_patcher(thread), caplog.at_level(logging.DEBUG): with daemon: @@ -266,18 +269,19 @@ class TestHttpHooksCallbackDaemon(object assert_message_in_log( caplog.records, 'Waiting for background thread to finish.', - levelno=logging.DEBUG, module='hooks_daemon') + levelno=logging.DEBUG, module='http_hooks_deamon') def _tcp_patcher(self, tcp_server): return mock.patch.object( - hooks_daemon, 'TCPServer', return_value=tcp_server) + http_hooks_deamon, 'TCPServer', return_value=tcp_server) def _thread_patcher(self, thread): return mock.patch.object( - hooks_daemon.threading, 'Thread', return_value=thread) + http_hooks_deamon.threading, 'Thread', return_value=thread) class TestPrepareHooksDaemon(object): + @pytest.mark.parametrize('protocol', ('celery',)) def test_returns_celery_hooks_callback_daemon_when_celery_protocol_specified( self, protocol): @@ -286,12 +290,12 @@ class TestPrepareHooksDaemon(object): "celery.result_backend = redis://redis/0") temp_file.flush() expected_extras = {'config': temp_file.name} - callback, extras = hooks_daemon.prepare_callback_daemon( + callback, extras = hook_base.prepare_callback_daemon( expected_extras, protocol=protocol, host='') - assert isinstance(callback, hooks_daemon.CeleryHooksCallbackDaemon) + assert isinstance(callback, celery_hooks_deamon.CeleryHooksCallbackDaemon) @pytest.mark.parametrize('protocol, expected_class', ( - ('http', hooks_daemon.HttpHooksCallbackDaemon), + ('http', http_hooks_deamon.HttpHooksCallbackDaemon), )) def test_returns_real_hooks_callback_daemon_when_protocol_is_specified( self, protocol, expected_class): @@ -302,7 +306,7 @@ class TestPrepareHooksDaemon(object): 'task_backend': '', 'task_queue': '' } - callback, extras = hooks_daemon.prepare_callback_daemon( + callback, extras = hook_base.prepare_callback_daemon( expected_extras.copy(), protocol=protocol, host='127.0.0.1', txn_id='txnid2') assert isinstance(callback, expected_class) @@ -321,7 +325,7 @@ class TestPrepareHooksDaemon(object): 'hooks_protocol': protocol.lower() } with pytest.raises(Exception): - callback, extras = hooks_daemon.prepare_callback_daemon( + callback, extras = hook_base.prepare_callback_daemon( expected_extras.copy(), protocol=protocol, host='127.0.0.1')