##// END OF EJS Templates
release: Bump version 5.1.0 to 5.1.1
release: Bump version 5.1.0 to 5.1.1

File last commit:

r1243:d32b737d default
r1269:20bfc127 v5.1.1 stable
Show More
test_hooks.py
257 lines | 7.9 KiB | text/x-python | PythonLexer
# 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 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
@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)
self.send_response(200)
self.end_headers()
self.wfile.write(body)
class MirrorHttpServer:
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)
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}