core: deprecated http based callback deamon.

It was now broken for locking/branch protection, and celery is stable enough to be ready as main and only backend

vcsserver/exceptions.py

@@ -1,131 +1,131 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation; either version 3 of the License, or
 # (at your option) any later version.
 #
 # 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 General Public License
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 
 """
 Special exception handling over the wire.
 
 Since we cannot assume that our client is able to import our exception classes,
 this module provides a "wrapping" mechanism to raise plain exceptions
 which contain an extra attribute `_vcs_kind` to allow a client to distinguish
 different error conditions.
 """
 
 from pyramid.httpexceptions import HTTPLocked, HTTPForbidden
 
 
 def _make_exception(kind, org_exc, *args):
     """
     Prepares a base `Exception` instance to be sent over the wire.
 
     To give our caller a hint what this is about, it will attach an attribute
     `_vcs_kind` to the exception.
     """
     exc = Exception(*args)
     exc._vcs_kind = kind
     exc._org_exc = org_exc
     exc._org_exc_tb = getattr(org_exc, '_org_exc_tb', '')
     return exc
 
 
 def AbortException(org_exc=None):
     def _make_exception_wrapper(*args):
         return _make_exception('abort', org_exc, *args)
     return _make_exception_wrapper
 
 
 def ArchiveException(org_exc=None):
     def _make_exception_wrapper(*args):
         return _make_exception('archive', org_exc, *args)
     return _make_exception_wrapper
 
 
 def ClientNotSupportedException(org_exc=None):
     def _make_exception_wrapper(*args):
         return _make_exception('client_not_supported', org_exc, *args)
     return _make_exception_wrapper
 
 
 def LookupException(org_exc=None):
     def _make_exception_wrapper(*args):
         return _make_exception('lookup', org_exc, *args)
     return _make_exception_wrapper
 
 
 def VcsException(org_exc=None):
     def _make_exception_wrapper(*args):
         return _make_exception('error', org_exc, *args)
     return _make_exception_wrapper
 
 
-def RepositoryLockedException(org_exc=None):
+def LockedRepoException(org_exc=None):
     def _make_exception_wrapper(*args):
         return _make_exception('repo_locked', org_exc, *args)
     return _make_exception_wrapper
 
 
 def RepositoryBranchProtectedException(org_exc=None):
     def _make_exception_wrapper(*args):
         return _make_exception('repo_branch_protected', org_exc, *args)
     return _make_exception_wrapper
 
 
 def RequirementException(org_exc=None):
     def _make_exception_wrapper(*args):
         return _make_exception('requirement', org_exc, *args)
     return _make_exception_wrapper
 
 
 def UnhandledException(org_exc=None):
     def _make_exception_wrapper(*args):
         return _make_exception('unhandled', org_exc, *args)
     return _make_exception_wrapper
 
 
 def URLError(org_exc=None):
     def _make_exception_wrapper(*args):
         return _make_exception('url_error', org_exc, *args)
     return _make_exception_wrapper
 
 
 def SubrepoMergeException(org_exc=None):
     def _make_exception_wrapper(*args):
         return _make_exception('subrepo_merge_error', org_exc, *args)
     return _make_exception_wrapper
 
 
 class HTTPRepoLocked(HTTPLocked):
     """
     Subclass of HTTPLocked response that allows to set the title and status
     code via constructor arguments.
     """
     def __init__(self, title, status_code=None, **kwargs):
         self.code = status_code or HTTPLocked.code
         self.title = title
         super().__init__(**kwargs)
 
 
 class HTTPRepoBranchProtected(HTTPForbidden):
     def __init__(self, *args, **kwargs):
         super(HTTPForbidden, self).__init__(*args, **kwargs)
 
 
 class RefNotFoundException(KeyError):
     pass
 
 
 class NoContentException(ValueError):
     pass

vcsserver/hooks.py

@@ -1,828 +1,780 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation; either version 3 of the License, or
 # (at your option) any later version.
 #
 # 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 General Public License
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 
 import io
 import os
 import sys
 import logging
 import collections
 import base64
 import msgpack
 import dataclasses
 import pygit2
 
 import http.client
 from celery import Celery
 
 import mercurial.scmutil
 import mercurial.node
 
 from vcsserver import exceptions, subprocessio, settings
 from vcsserver.lib.ext_json import json
 from vcsserver.lib.str_utils import ascii_str, safe_str
 from vcsserver.lib.svn_txn_utils import get_txn_id_from_store
 from vcsserver.remote.git_remote import Repository
 
 celery_app = Celery('__vcsserver__')
 log = logging.getLogger(__name__)
 
 
-class HooksHttpClient:
-    proto = 'msgpack.v1'
-    connection = None
-
-    def __init__(self, hooks_uri):
-        self.hooks_uri = hooks_uri
-
-    def __repr__(self):
-        return f'{self.__class__}(hook_uri={self.hooks_uri}, proto={self.proto})'
-
-    def __call__(self, method, extras):
-        connection = http.client.HTTPConnection(self.hooks_uri)
-        # binary msgpack body
-        headers, body = self._serialize(method, extras)
-        log.debug('Doing a new hooks call using HTTPConnection to %s', self.hooks_uri)
-
-        try:
-            try:
-                connection.request('POST', '/', body, headers)
-            except Exception as error:
-                log.error('Hooks calling Connection failed on %s, org error: %s', connection.__dict__, error)
-                raise
-
-            response = connection.getresponse()
-            try:
-                return msgpack.load(response)
-            except Exception:
-                response_data = response.read()
-                log.exception('Failed to decode hook response json data. '
-                              'response_code:%s, raw_data:%s',
-                              response.status, response_data)
-                raise
-        finally:
-            connection.close()
-
-    @classmethod
-    def _serialize(cls, hook_name, extras):
-        data = {
-            'method': hook_name,
-            'extras': extras
-        }
-        headers = {
-            "rc-hooks-protocol": cls.proto,
-            "Connection": "keep-alive"
-        }
-        return headers, msgpack.packb(data)
-
-
 class HooksCeleryClient:
     TASK_TIMEOUT = 60  # time in seconds
 
     def __init__(self, queue, backend):
         celery_app.config_from_object({
             'broker_url': queue, 'result_backend': backend,
             'broker_connection_retry_on_startup': True,
             'task_serializer': 'json',
             'accept_content': ['json', 'msgpack'],
             'result_serializer': 'json',
             'result_accept_content': ['json', 'msgpack']
         })
         self.celery_app = celery_app
 
     def __call__(self, method, extras):
         inquired_task = self.celery_app.signature(
             f'rhodecode.lib.celerylib.tasks.{method}'
         )
         result = inquired_task.delay(extras).get(timeout=self.TASK_TIMEOUT)
 
         return result
 
 
 class HooksShadowRepoClient:
 
     def __call__(self, hook_name, extras):
         return {'output': '', 'status': 0}
 
 
 class RemoteMessageWriter:
     """Writer base class."""
     def write(self, message):
         raise NotImplementedError()
 
 
 class HgMessageWriter(RemoteMessageWriter):
     """Writer that knows how to send messages to mercurial clients."""
 
     def __init__(self, ui):
         self.ui = ui
 
     def write(self, message: str):
         # TODO: Check why the quiet flag is set by default.
         old = self.ui.quiet
         self.ui.quiet = False
         self.ui.status(message.encode('utf-8'))
         self.ui.quiet = old
 
 
 class GitMessageWriter(RemoteMessageWriter):
     """Writer that knows how to send messages to git clients."""
 
     def __init__(self, stdout=None):
         self.stdout = stdout or sys.stdout
 
     def write(self, message: str):
         self.stdout.write(message)
 
 
 class SvnMessageWriter(RemoteMessageWriter):
     """Writer that knows how to send messages to svn clients."""
 
     def __init__(self, stderr=None):
         # SVN needs data sent to stderr for back-to-client messaging
         self.stderr = stderr or sys.stderr
 
     def write(self, message):
         self.stderr.write(message)
 
 
 def _maybe_handle_exception(result):
+
+
     exception_class = result.get('exception')
     exception_traceback = result.get('exception_traceback')
-    if not (exception_class and exception_traceback):
+    if not exception_class:
         return
+
     log.debug('Handling hook-call exception: %s', exception_class)
 
     if exception_traceback:
         log.error('Got traceback from remote call:%s', exception_traceback)
 
-    if exception_class == 'HTTPLockedRC':
-        raise exceptions.RepositoryLockedException()(*result['exception_args'])
+    if exception_class == 'HTTPLockedRepo':
+        raise exceptions.LockedRepoException()(*result['exception_args'])
     elif exception_class == 'ClientNotSupportedError':
         raise exceptions.ClientNotSupportedException()(*result['exception_args'])
     elif exception_class == 'HTTPBranchProtected':
         raise exceptions.RepositoryBranchProtectedException()(*result['exception_args'])
     elif exception_class == 'RepositoryError':
         raise exceptions.VcsException()(*result['exception_args'])
     elif exception_class:
         raise Exception(
             f"""Got remote exception "{exception_class}" with args "{result['exception_args']}" """
         )
 
 
 def _get_hooks_client(extras):
-    hooks_uri = extras.get('hooks_uri')
     task_queue = extras.get('task_queue')
     task_backend = extras.get('task_backend')
     is_shadow_repo = extras.get('is_shadow_repo')
 
-    if hooks_uri:
-        return HooksHttpClient(hooks_uri)
-    elif task_queue and task_backend:
+    if task_queue and task_backend:
         return HooksCeleryClient(task_queue, task_backend)
     elif is_shadow_repo:
         return HooksShadowRepoClient()
     else:
         raise Exception("Hooks client not found!")
 
 
 def _call_hook(hook_name, extras, writer):
     hooks_client = _get_hooks_client(extras)
     log.debug('Hooks, using client:%s', hooks_client)
     result = hooks_client(hook_name, extras)
     log.debug('Hooks got result: %s', result)
     _maybe_handle_exception(result)
     writer.write(result['output'])
 
     return result['status']
 
 
 def _extras_from_ui(ui):
     hook_data = ui.config(b'rhodecode', b'RC_SCM_DATA')
     if not hook_data:
         # maybe it's inside environ ?
         env_hook_data = os.environ.get('RC_SCM_DATA')
         if env_hook_data:
             hook_data = env_hook_data
 
     extras = {}
     if hook_data:
         extras = json.loads(hook_data)
     return extras
 
 
 def _rev_range_hash(repo, node, check_heads=False):
     from vcsserver.hgcompat import get_ctx
 
     commits = []
     revs = []
     start = get_ctx(repo, node).rev()
     end = len(repo)
     for rev in range(start, end):
         revs.append(rev)
         ctx = get_ctx(repo, rev)
         commit_id = ascii_str(mercurial.node.hex(ctx.node()))
         branch = safe_str(ctx.branch())
         commits.append((commit_id, branch))
 
     parent_heads = []
     if check_heads:
         parent_heads = _check_heads(repo, start, end, revs)
     return commits, parent_heads
 
 
 def _check_heads(repo, start, end, commits):
     from vcsserver.hgcompat import get_ctx
     changelog = repo.changelog
     parents = set()
 
     for new_rev in commits:
         for p in changelog.parentrevs(new_rev):
             if p == mercurial.node.nullrev:
                 continue
             if p < start:
                 parents.add(p)
 
     for p in parents:
         branch = get_ctx(repo, p).branch()
         # The heads descending from that parent, on the same branch
         parent_heads = {p}
         reachable = {p}
         for x in range(p + 1, end):
             if get_ctx(repo, x).branch() != branch:
                 continue
             for pp in changelog.parentrevs(x):
                 if pp in reachable:
                     reachable.add(x)
                     parent_heads.discard(pp)
                     parent_heads.add(x)
         # More than one head? Suggest merging
         if len(parent_heads) > 1:
             return list(parent_heads)
 
     return []
 
 
 def _get_git_env():
     env = {}
     for k, v in os.environ.items():
         if k.startswith('GIT'):
             env[k] = v
 
     # serialized version
     return [(k, v) for k, v in env.items()]
 
 
 def _get_hg_env(old_rev, new_rev, txnid, repo_path):
     env = {}
     for k, v in os.environ.items():
         if k.startswith('HG'):
             env[k] = v
 
     env['HG_NODE'] = old_rev
     env['HG_NODE_LAST'] = new_rev
     env['HG_TXNID'] = txnid
     env['HG_PENDING'] = repo_path
 
     return [(k, v) for k, v in env.items()]
 
 
 def _get_ini_settings(ini_file):
     from vcsserver.http_main import sanitize_settings_and_apply_defaults
     from vcsserver.lib.config_utils import get_app_config_lightweight, configure_and_store_settings
 
     global_config = {'__file__': ini_file}
     ini_settings = get_app_config_lightweight(ini_file)
     sanitize_settings_and_apply_defaults(global_config, ini_settings)
     configure_and_store_settings(global_config, ini_settings)
 
     return ini_settings
 
 
 def _fix_hooks_executables(ini_path=''):
     """
     This is a trick to set proper settings.EXECUTABLE paths for certain execution patterns
     especially for subversion where hooks strip entire env, and calling just 'svn' command will most likely fail
     because svn is not on PATH
     """
     # set defaults, in case we can't read from ini_file
     core_binary_dir = settings.BINARY_DIR or '/usr/local/bin/rhodecode_bin/vcs_bin'
     if ini_path:
         ini_settings = _get_ini_settings(ini_path)
         core_binary_dir = ini_settings['core.binary_dir']
 
     settings.BINARY_DIR = core_binary_dir
 
 
 def repo_size(ui, repo, **kwargs):
     extras = _extras_from_ui(ui)
     return _call_hook('repo_size', extras, HgMessageWriter(ui))
 
 
 def pre_pull(ui, repo, **kwargs):
     extras = _extras_from_ui(ui)
     return _call_hook('pre_pull', extras, HgMessageWriter(ui))
 
 
 def pre_pull_ssh(ui, repo, **kwargs):
     extras = _extras_from_ui(ui)
     if extras and extras.get('SSH'):
         return pre_pull(ui, repo, **kwargs)
     return 0
 
 
 def post_pull(ui, repo, **kwargs):
     extras = _extras_from_ui(ui)
     return _call_hook('post_pull', extras, HgMessageWriter(ui))
 
 
 def post_pull_ssh(ui, repo, **kwargs):
     extras = _extras_from_ui(ui)
     if extras and extras.get('SSH'):
         return post_pull(ui, repo, **kwargs)
     return 0
 
 
 def pre_push(ui, repo, node=None, **kwargs):
     """
     Mercurial pre_push hook
     """
     extras = _extras_from_ui(ui)
     detect_force_push = extras.get('detect_force_push')
 
     rev_data = []
     hook_type: str = safe_str(kwargs.get('hooktype'))
 
     if node and hook_type == 'pretxnchangegroup':
         branches = collections.defaultdict(list)
         commits, _heads = _rev_range_hash(repo, node, check_heads=detect_force_push)
         for commit_id, branch in commits:
             branches[branch].append(commit_id)
 
         for branch, commits in branches.items():
             old_rev = ascii_str(kwargs.get('node_last')) or commits[0]
             rev_data.append({
                 'total_commits': len(commits),
                 'old_rev': old_rev,
                 'new_rev': commits[-1],
                 'ref': '',
                 'type': 'branch',
                 'name': branch,
             })
 
         for push_ref in rev_data:
             push_ref['multiple_heads'] = _heads
 
             repo_path = os.path.join(
                 extras.get('repo_store', ''), extras.get('repository', ''))
             push_ref['hg_env'] = _get_hg_env(
                 old_rev=push_ref['old_rev'],
                 new_rev=push_ref['new_rev'], txnid=ascii_str(kwargs.get('txnid')),
                 repo_path=repo_path)
 
     extras['hook_type'] = hook_type or 'pre_push'
     extras['commit_ids'] = rev_data
 
     return _call_hook('pre_push', extras, HgMessageWriter(ui))
 
 
 def pre_push_ssh(ui, repo, node=None, **kwargs):
     extras = _extras_from_ui(ui)
     if extras.get('SSH'):
         return pre_push(ui, repo, node, **kwargs)
 
     return 0
 
 
 def pre_push_ssh_auth(ui, repo, node=None, **kwargs):
     """
     Mercurial pre_push hook for SSH
     """
     extras = _extras_from_ui(ui)
     if extras.get('SSH'):
         permission = extras['SSH_PERMISSIONS']
 
         if 'repository.write' == permission or 'repository.admin' == permission:
             return 0
 
         # non-zero ret code
         return 1
 
     return 0
 
 
 def post_push(ui, repo, node, **kwargs):
     """
     Mercurial post_push hook
     """
     extras = _extras_from_ui(ui)
 
     commit_ids = []
     branches = []
     bookmarks = []
     tags = []
     hook_type: str = safe_str(kwargs.get('hooktype'))
 
     commits, _heads = _rev_range_hash(repo, node)
     for commit_id, branch in commits:
         commit_ids.append(commit_id)
         if branch not in branches:
             branches.append(branch)
 
     if hasattr(ui, '_rc_pushkey_bookmarks'):
         bookmarks = ui._rc_pushkey_bookmarks
 
     extras['hook_type'] = hook_type or 'post_push'
     extras['commit_ids'] = commit_ids
 
     extras['new_refs'] = {
         'branches': branches,
         'bookmarks': bookmarks,
         'tags': tags
     }
 
     return _call_hook('post_push', extras, HgMessageWriter(ui))
 
 
 def post_push_ssh(ui, repo, node, **kwargs):
     """
     Mercurial post_push hook for SSH
     """
     if _extras_from_ui(ui).get('SSH'):
         return post_push(ui, repo, node, **kwargs)
     return 0
 
 
 def key_push(ui, repo, **kwargs):
     from vcsserver.hgcompat import get_ctx
 
     if kwargs['new'] != b'0' and kwargs['namespace'] == b'bookmarks':
         # store new bookmarks in our UI object propagated later to post_push
         ui._rc_pushkey_bookmarks = get_ctx(repo, kwargs['key']).bookmarks()
     return
 
 
 # backward compat
 log_pull_action = post_pull
 
 # backward compat
 log_push_action = post_push
 
 
 def handle_git_pre_receive(unused_repo_path, unused_revs, unused_env):
     """
     Old hook name: keep here for backward compatibility.
 
     This is only required when the installed git hooks are not upgraded.
     """
     pass
 
 
 def handle_git_post_receive(unused_repo_path, unused_revs, unused_env):
     """
     Old hook name: keep here for backward compatibility.
 
     This is only required when the installed git hooks are not upgraded.
     """
     pass
 
 
 @dataclasses.dataclass
 class HookResponse:
     status: int
     output: str
 
 
 def git_pre_pull(extras) -> HookResponse:
     """
     Pre pull hook.
 
     :param extras: dictionary containing the keys defined in simplevcs
     :type extras: dict
 
     :return: status code of the hook. 0 for success.
     :rtype: int
     """
 
     if 'pull' not in extras['hooks']:
         return HookResponse(0, '')
 
     stdout = io.StringIO()
     try:
         status_code = _call_hook('pre_pull', extras, GitMessageWriter(stdout))
 
     except Exception as error:
         log.exception('Failed to call pre_pull hook')
         status_code = 128
         stdout.write(f'ERROR: {error}\n')
 
     return HookResponse(status_code, stdout.getvalue())
 
 
 def git_post_pull(extras) -> HookResponse:
     """
     Post pull hook.
 
     :param extras: dictionary containing the keys defined in simplevcs
     :type extras: dict
 
     :return: status code of the hook. 0 for success.
     :rtype: int
     """
     if 'pull' not in extras['hooks']:
         return HookResponse(0, '')
 
     stdout = io.StringIO()
     try:
         status = _call_hook('post_pull', extras, GitMessageWriter(stdout))
     except Exception as error:
         status = 128
         stdout.write(f'ERROR: {error}\n')
 
     return HookResponse(status, stdout.getvalue())
 
 
 def _parse_git_ref_lines(revision_lines):
     rev_data = []
     for revision_line in revision_lines or []:
         old_rev, new_rev, ref = revision_line.strip().split(' ')
         ref_data = ref.split('/', 2)
         if ref_data[1] in ('tags', 'heads'):
             rev_data.append({
                 # NOTE(marcink):
                 # we're unable to tell total_commits for git at this point
                 # but we set the variable for consistency with GIT
                 'total_commits': -1,
                 'old_rev': old_rev,
                 'new_rev': new_rev,
                 'ref': ref,
                 'type': ref_data[1],
                 'name': ref_data[2],
             })
     return rev_data
 
 
 def git_pre_receive(unused_repo_path, revision_lines, env) -> int:
     """
     Pre push hook.
 
     :return: status code of the hook. 0 for success.
     """
     extras = json.loads(env['RC_SCM_DATA'])
     rev_data = _parse_git_ref_lines(revision_lines)
     if 'push' not in extras['hooks']:
         return 0
     _fix_hooks_executables(env.get('RC_INI_FILE'))
 
     empty_commit_id = '0' * 40
 
     detect_force_push = extras.get('detect_force_push')
 
     for push_ref in rev_data:
         # store our git-env which holds the temp store
         push_ref['git_env'] = _get_git_env()
         push_ref['pruned_sha'] = ''
         if not detect_force_push:
             # don't check for forced-push when we don't need to
             continue
 
         type_ = push_ref['type']
         new_branch = push_ref['old_rev'] == empty_commit_id
         delete_branch = push_ref['new_rev'] == empty_commit_id
         if type_ == 'heads' and not (new_branch or delete_branch):
             old_rev = push_ref['old_rev']
             new_rev = push_ref['new_rev']
             cmd = [settings.GIT_EXECUTABLE(), 'rev-list', old_rev, f'^{new_rev}']
             stdout, stderr = subprocessio.run_command(
                 cmd, env=os.environ.copy())
             # means we're having some non-reachable objects, this forced push was used
             if stdout:
                 push_ref['pruned_sha'] = stdout.splitlines()
 
     extras['hook_type'] = 'pre_receive'
     extras['commit_ids'] = rev_data
 
     stdout = sys.stdout
     status_code = _call_hook('pre_push', extras, GitMessageWriter(stdout))
 
     return status_code
 
 
 def git_post_receive(unused_repo_path, revision_lines, env) -> int:
     """
     Post push hook.
 
     :return: status code of the hook. 0 for success.
     """
     extras = json.loads(env['RC_SCM_DATA'])
     if 'push' not in extras['hooks']:
         return 0
 
     _fix_hooks_executables(env.get('RC_INI_FILE'))
 
     rev_data = _parse_git_ref_lines(revision_lines)
 
     git_revs = []
 
     # N.B.(skreft): it is ok to just call git, as git before calling a
     # subcommand sets the PATH environment variable so that it point to the
     # correct version of the git executable.
     empty_commit_id = '0' * 40
     branches = []
     tags = []
     for push_ref in rev_data:
         type_ = push_ref['type']
 
         if type_ == 'heads':
             # starting new branch case
             if push_ref['old_rev'] == empty_commit_id:
                 push_ref_name = push_ref['name']
 
                 if push_ref_name not in branches:
                     branches.append(push_ref_name)
 
                 need_head_set = ''
                 with Repository(os.getcwd()) as repo:
                     try:
                         repo.head
                     except pygit2.GitError:
                         need_head_set = f'refs/heads/{push_ref_name}'
 
                     if need_head_set:
                         repo.set_head(need_head_set)
                         print(f"Setting default branch to {push_ref_name}")
 
                 cmd = [settings.GIT_EXECUTABLE(), 'for-each-ref', '--format=%(refname)', 'refs/heads/*']
                 stdout, stderr = subprocessio.run_command(
                     cmd, env=os.environ.copy())
                 heads = safe_str(stdout)
                 heads = heads.replace(push_ref['ref'], '')
                 heads = ' '.join(head for head
                                  in heads.splitlines() if head) or '.'
                 cmd = [settings.GIT_EXECUTABLE(), 'log', '--reverse',
                        '--pretty=format:%H', '--', push_ref['new_rev'],
                        '--not', heads]
                 stdout, stderr = subprocessio.run_command(
                     cmd, env=os.environ.copy())
                 git_revs.extend(list(map(ascii_str, stdout.splitlines())))
 
             # delete branch case
             elif push_ref['new_rev'] == empty_commit_id:
                 git_revs.append(f'delete_branch=>{push_ref["name"]}')
             else:
                 if push_ref['name'] not in branches:
                     branches.append(push_ref['name'])
 
                 cmd = [settings.GIT_EXECUTABLE(), 'log',
                        f'{push_ref["old_rev"]}..{push_ref["new_rev"]}',
                        '--reverse', '--pretty=format:%H']
                 stdout, stderr = subprocessio.run_command(
                     cmd, env=os.environ.copy())
                 # we get bytes from stdout, we need str to be consistent
                 log_revs = list(map(ascii_str, stdout.splitlines()))
                 git_revs.extend(log_revs)
 
                 # Pure pygit2 impl. but still 2-3x slower :/
                 # results = []
                 #
                 # with Repository(os.getcwd()) as repo:
                 #     repo_new_rev = repo[push_ref['new_rev']]
                 #     repo_old_rev = repo[push_ref['old_rev']]
                 #     walker = repo.walk(repo_new_rev.id, pygit2.GIT_SORT_TOPOLOGICAL)
                 #
                 #     for commit in walker:
                 #         if commit.id == repo_old_rev.id:
                 #             break
                 #         results.append(commit.id.hex)
                 #     # reverse the order, can't use GIT_SORT_REVERSE
                 #     log_revs = results[::-1]
 
         elif type_ == 'tags':
             if push_ref['name'] not in tags:
                 tags.append(push_ref['name'])
             git_revs.append(f'tag=>{push_ref["name"]}')
 
     extras['hook_type'] = 'post_receive'
     extras['commit_ids'] = git_revs
     extras['new_refs'] = {
         'branches': branches,
         'bookmarks': [],
         'tags': tags,
     }
 
     stdout = sys.stdout
 
     if 'repo_size' in extras['hooks']:
         try:
             _call_hook('repo_size', extras, GitMessageWriter(stdout))
         except Exception:
             pass
 
     status_code = _call_hook('post_push', extras, GitMessageWriter(stdout))
     return status_code
 
 
 def get_extras_from_txn_id(repo_path, txn_id):
     extras = get_txn_id_from_store(repo_path, txn_id)
     return extras
 
 
 def svn_pre_commit(repo_path, commit_data, env):
 
     path, txn_id = commit_data
     branches = []
     tags = []
 
     if env.get('RC_SCM_DATA'):
         extras = json.loads(env['RC_SCM_DATA'])
     else:
         ini_path = env.get('RC_INI_FILE')
         if ini_path:
             _get_ini_settings(ini_path)
         # fallback method to read from TXN-ID stored data
         extras = get_extras_from_txn_id(path, txn_id)
 
     if not extras:
         raise ValueError('SVN-PRE-COMMIT: Failed to extract context data in called extras for hook execution')
 
     if extras.get('rc_internal_commit'):
         # special marker for internal commit, we don't call hooks client
         return 0
 
     extras['hook_type'] = 'pre_commit'
     extras['commit_ids'] = [txn_id]
     extras['txn_id'] = txn_id
     extras['new_refs'] = {
         'total_commits': 1,
         'branches': branches,
         'bookmarks': [],
         'tags': tags,
     }
 
     return _call_hook('pre_push', extras, SvnMessageWriter())
 
 
 def svn_post_commit(repo_path, commit_data, env):
     """
     commit_data is path, rev, txn_id
     """
 
     if len(commit_data) == 3:
         path, commit_id, txn_id = commit_data
     elif len(commit_data) == 2:
         log.error('Failed to extract txn_id from commit_data using legacy method. '
                   'Some functionality might be limited')
         path, commit_id = commit_data
         txn_id = None
     else:
         return 0
 
     branches = []
     tags = []
 
     if env.get('RC_SCM_DATA'):
         extras = json.loads(env['RC_SCM_DATA'])
     else:
         ini_path = env.get('RC_INI_FILE')
         if ini_path:
             _get_ini_settings(ini_path)
         # fallback method to read from TXN-ID stored data
         extras = get_extras_from_txn_id(path, txn_id)
 
     if not extras and txn_id:
         raise ValueError('SVN-POST-COMMIT: Failed to extract context data in called extras for hook execution')
 
     if extras.get('rc_internal_commit'):
         # special marker for internal commit, we don't call hooks client
         return 0
 
     extras['hook_type'] = 'post_commit'
     extras['commit_ids'] = [commit_id]
     extras['txn_id'] = txn_id
     extras['new_refs'] = {
         'branches': branches,
         'bookmarks': [],
         'tags': tags,
         'total_commits': 1,
     }
 
     if 'repo_size' in extras['hooks']:
         try:
             _call_hook('repo_size', extras, SvnMessageWriter())
         except Exception:
             pass
 
     return _call_hook('post_push', extras, SvnMessageWriter())

vcsserver/tests/test_hooks.py

@@ -1,257 +1,135 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation; either version 3 of the License, or
 # (at your option) any later version.
 #
 # 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 General Public License
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 
-import threading
-import msgpack
-
-from http.server import BaseHTTPRequestHandler
-from socketserver import TCPServer
+import pytest
 
 import mercurial.ui
 import mock
-import pytest
 
-from vcsserver.hooks import HooksHttpClient
 from vcsserver.lib.ext_json import json
 from vcsserver import hooks
 
 
 def get_hg_ui(extras=None):
     """Create a Config object with a valid RC_SCM_DATA entry."""
     extras = extras or {}
     required_extras = {
         'username': '',
         'repository': '',
         'locked_by': '',
         'scm': '',
         'make_lock': '',
         'action': '',
         'ip': '',
-        'hooks_uri': 'fake_hooks_uri',
     }
     required_extras.update(extras)
     hg_ui = mercurial.ui.ui()
     hg_ui.setconfig(b'rhodecode', b'RC_SCM_DATA', json.dumps(required_extras))
 
     return hg_ui
 
 
 def test_git_pre_receive_is_disabled():
     extras = {'hooks': ['pull']}
     response = hooks.git_pre_receive(None, None,
                                      {'RC_SCM_DATA': json.dumps(extras)})
 
     assert response == 0
 
 
 def test_git_post_receive_is_disabled():
     extras = {'hooks': ['pull']}
     response = hooks.git_post_receive(None, '',
                                       {'RC_SCM_DATA': json.dumps(extras)})
 
     assert response == 0
 
 
 def test_git_post_receive_calls_repo_size():
     extras = {'hooks': ['push', 'repo_size']}
 
     with mock.patch.object(hooks, '_call_hook') as call_hook_mock:
         hooks.git_post_receive(
             None, '', {'RC_SCM_DATA': json.dumps(extras)})
     extras.update({'commit_ids': [], 'hook_type': 'post_receive',
                    'new_refs': {'bookmarks': [], 'branches': [], 'tags': []}})
     expected_calls = [
         mock.call('repo_size', extras, mock.ANY),
         mock.call('post_push', extras, mock.ANY),
     ]
     assert call_hook_mock.call_args_list == expected_calls
 
 
 def test_git_post_receive_does_not_call_disabled_repo_size():
     extras = {'hooks': ['push']}
 
     with mock.patch.object(hooks, '_call_hook') as call_hook_mock:
         hooks.git_post_receive(
             None, '', {'RC_SCM_DATA': json.dumps(extras)})
     extras.update({'commit_ids': [], 'hook_type': 'post_receive',
                    'new_refs': {'bookmarks': [], 'branches': [], 'tags': []}})
     expected_calls = [
         mock.call('post_push', extras, mock.ANY)
     ]
     assert call_hook_mock.call_args_list == expected_calls
 
 
 def test_repo_size_exception_does_not_affect_git_post_receive():
     extras = {'hooks': ['push', 'repo_size']}
     status = 0
 
     def side_effect(name, *args, **kwargs):
         if name == 'repo_size':
             raise Exception('Fake exception')
         else:
             return status
 
     with mock.patch.object(hooks, '_call_hook') as call_hook_mock:
         call_hook_mock.side_effect = side_effect
         result = hooks.git_post_receive(
             None, '', {'RC_SCM_DATA': json.dumps(extras)})
     assert result == status
 
 
 def test_git_pre_pull_is_disabled():
     assert hooks.git_pre_pull({'hooks': ['push']}) == hooks.HookResponse(0, '')
 
 
 def test_git_post_pull_is_disabled():
     assert (
         hooks.git_post_pull({'hooks': ['push']}) == hooks.HookResponse(0, ''))
 
 
 class TestGetHooksClient:
 
-    def test_returns_http_client_when_protocol_matches(self):
-        hooks_uri = 'localhost:8000'
-        result = hooks._get_hooks_client({
-            'hooks_uri': hooks_uri,
-            'hooks_protocol': 'http'
-        })
-        assert isinstance(result, hooks.HooksHttpClient)
-        assert result.hooks_uri == hooks_uri
-
     def test_return_celery_client_when_queue_and_backend_provided(self):
         task_queue = 'redis://task_queue:0'
         task_backend = task_queue
         result = hooks._get_hooks_client({
             'task_queue': task_queue,
             'task_backend': task_backend
         })
         assert isinstance(result, hooks.HooksCeleryClient)
 
 
-class TestHooksHttpClient:
-    def test_init_sets_hooks_uri(self):
-        uri = 'localhost:3000'
-        client = hooks.HooksHttpClient(uri)
-        assert client.hooks_uri == uri
-
-    def test_serialize_returns_serialized_string(self):
-        client = hooks.HooksHttpClient('localhost:3000')
-        hook_name = 'test'
-        extras = {
-            'first': 1,
-            'second': 'two'
-        }
-        hooks_proto, result = client._serialize(hook_name, extras)
-        expected_result = msgpack.packb({
-            'method': hook_name,
-            'extras': extras,
-        })
-        assert hooks_proto == {'rc-hooks-protocol': 'msgpack.v1', 'Connection': 'keep-alive'}
-        assert result == expected_result
-
-    def test_call_queries_http_server(self, http_mirror):
-        client = hooks.HooksHttpClient(http_mirror.uri)
-        hook_name = 'test'
-        extras = {
-            'first': 1,
-            'second': 'two'
-        }
-        result = client(hook_name, extras)
-        expected_result = msgpack.unpackb(msgpack.packb({
-            'method': hook_name,
-            'extras': extras
-        }), raw=False)
-        assert result == expected_result
-
-
[e-mail]
-def http_mirror(request):
-    server = MirrorHttpServer()
-    request.addfinalizer(server.stop)
-    return server
-
-
-class MirrorHttpHandler(BaseHTTPRequestHandler):
-
-    def do_POST(self):
-        length = int(self.headers['Content-Length'])
-        body = self.rfile.read(length)
-        self.send_response(200)
-        self.end_headers()
-        self.wfile.write(body)
-
-
-class MirrorHttpServer:
-    ip_address = '127.0.0.1'
-    port = 0
+class TestHooksCeleryClient:
 
-    def __init__(self):
-        self._daemon = TCPServer((self.ip_address, 0), MirrorHttpHandler)
-        _, self.port = self._daemon.server_address
-        self._thread = threading.Thread(target=self._daemon.serve_forever)
-        self._thread.daemon = True
-        self._thread.start()
-
-    def stop(self):
-        self._daemon.shutdown()
-        self._thread.join()
-        self._daemon = None
-        self._thread = None
-
-    @property
-    def uri(self):
-        return '{}:{}'.format(self.ip_address, self.port)
-
-
-def test_hooks_http_client_init():
-    hooks_uri = 'http://localhost:8000'
-    client = HooksHttpClient(hooks_uri)
-    assert client.hooks_uri == hooks_uri
-
-
-def test_hooks_http_client_call():
-    hooks_uri = 'http://localhost:8000'
-
-    method = 'test_method'
-    extras = {'key': 'value'}
-
-    with \
-        mock.patch('http.client.HTTPConnection') as mock_connection,\
-        mock.patch('msgpack.load') as mock_load:
-
-        client = HooksHttpClient(hooks_uri)
-
-        mock_load.return_value = {'result': 'success'}
-        response = mock.MagicMock()
-        response.status = 200
-        mock_connection.request.side_effect = None
-        mock_connection.getresponse.return_value = response
-
-        result = client(method, extras)
-
-        mock_connection.assert_called_with(hooks_uri)
-        mock_connection.return_value.request.assert_called_once()
-        assert result == {'result': 'success'}
-
-
-def test_hooks_http_client_serialize():
-    method = 'test_method'
-    extras = {'key': 'value'}
-    headers, body = HooksHttpClient._serialize(method, extras)
-
-    assert headers == {'rc-hooks-protocol': HooksHttpClient.proto, 'Connection': 'keep-alive'}
-    assert msgpack.unpackb(body) == {'method': method, 'extras': extras}
+    def test_hooks_http_client_init(self):
+        queue = 'redis://redis:6379/0'
+        backend = 'redis://redis:6379/0'
+        client = hooks.HooksCeleryClient(queue, backend)
+        assert client.celery_app.conf.broker_url == queue