vcsserver/tests/test_hooks.py

# RhodeCode VCSServer provides access to different vcs backends via network.
# Copyright (C) 2014-2020 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 contextlib
import io
import threading
from BaseHTTPServer import BaseHTTPRequestHandler
from SocketServer import TCPServer

import mercurial.ui
import mock
import pytest
import simplejson as 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('rhodecode', '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(object):

    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_returns_dummy_client_when_hooks_uri_not_specified(self):
        fake_module = mock.Mock()
        import_patcher = mock.patch.object(
            hooks.importlib, 'import_module', return_value=fake_module)
        fake_module_name = 'fake.module'
        with import_patcher as import_mock:
            result = hooks._get_hooks_client(
                {'hooks_module': fake_module_name})

        import_mock.assert_called_once_with(fake_module_name)
        assert isinstance(result, hooks.HooksDummyClient)
        assert result._hooks_module == fake_module


class TestHooksHttpClient(object):
    def test_init_sets_hooks_uri(self):
        uri = 'localhost:3000'
        client = hooks.HooksHttpClient(uri)
        assert client.hooks_uri == uri

    def test_serialize_returns_json_string(self):
        client = hooks.HooksHttpClient('localhost:3000')
        hook_name = 'test'
        extras = {
            'first': 1,
            'second': 'two'
        }
        result = client._serialize(hook_name, extras)
        expected_result = json.dumps({
            'method': hook_name,
            'extras': extras
        })
        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 = {
            'method': hook_name,
            'extras': extras
        }
        assert result == expected_result


class TestHooksDummyClient(object):
    def test_init_imports_hooks_module(self):
        hooks_module_name = 'rhodecode.fake.module'
        hooks_module = mock.MagicMock()

        import_patcher = mock.patch.object(
            hooks.importlib, 'import_module', return_value=hooks_module)
        with import_patcher as import_mock:
            client = hooks.HooksDummyClient(hooks_module_name)
            import_mock.assert_called_once_with(hooks_module_name)
        assert client._hooks_module == hooks_module

    def test_call_returns_hook_result(self):
        hooks_module_name = 'rhodecode.fake.module'
        hooks_module = mock.MagicMock()
        import_patcher = mock.patch.object(
            hooks.importlib, 'import_module', return_value=hooks_module)
        with import_patcher:
            client = hooks.HooksDummyClient(hooks_module_name)

        result = client('post_push', {})
        hooks_module.Hooks.assert_called_once_with()
        assert result == hooks_module.Hooks().__enter__().post_push()


@pytest.fixture
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).decode('utf-8')
        self.send_response(200)
        self.end_headers()
        self.wfile.write(body)


class MirrorHttpServer(object):
    ip_address = '127.0.0.1'
    port = 0

    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)