##// END OF EJS Templates
fix(git): make the fetch revs use smaller revs to be less memory hungry, and less prone to fail
fix(git): make the fetch revs use smaller revs to be less memory hungry, and less prone to fail

File last commit:

r1152:a0c49580 default
r1185:729a204e default
Show More
test_hooks.py
286 lines | 9.1 KiB | text/x-python | PythonLexer
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 # RhodeCode VCSServer provides access to different vcs backends via network.
source-code: updated copyrights to 2023
r1126 # Copyright (C) 2014-2023 RhodeCode GmbH
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 #
# 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
python3: code change for py3 support...
r1048 import msgpack
py3: import/code fixes
r987 from http.server import BaseHTTPRequestHandler
from socketserver import TCPServer
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130
import mercurial.ui
import mock
import pytest
tests: added few generated tests
r1087 from vcsserver.hooks import HooksHttpClient
python3: code change for py3 support...
r1048 from vcsserver.lib.rc_json import json
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 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()
python3: code change for py3 support...
r1048 hg_ui.setconfig(b'rhodecode', b'RC_SCM_DATA', json.dumps(required_extras))
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130
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']}
python3: code change for py3 support...
r1048
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 with mock.patch.object(hooks, '_call_hook') as call_hook_mock:
hooks.git_post_receive(
None, '', {'RC_SCM_DATA': json.dumps(extras)})
hooks: store hook type for extensions.
r555 extras.update({'commit_ids': [], 'hook_type': 'post_receive',
hooks: expose pushed refs inside hooks....
r223 'new_refs': {'bookmarks': [], 'branches': [], 'tags': []}})
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 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']}
python3: code change for py3 support...
r1048
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 with mock.patch.object(hooks, '_call_hook') as call_hook_mock:
hooks.git_post_receive(
None, '', {'RC_SCM_DATA': json.dumps(extras)})
hooks: store hook type for extensions.
r555 extras.update({'commit_ids': [], 'hook_type': 'post_receive',
hooks: expose pushed refs inside hooks....
r223 'new_refs': {'bookmarks': [], 'branches': [], 'tags': []}})
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 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, ''))
lint: auto-fixes
r1152 class TestGetHooksClient:
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130
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
lint: auto-fixes
r1152 class TestHooksHttpClient:
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 def test_init_sets_hooks_uri(self):
uri = 'localhost:3000'
client = hooks.HooksHttpClient(uri)
assert client.hooks_uri == uri
python3: code change for py3 support...
r1048 def test_serialize_returns_serialized_string(self):
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 client = hooks.HooksHttpClient('localhost:3000')
hook_name = 'test'
extras = {
'first': 1,
'second': 'two'
}
python3: code change for py3 support...
r1048 hooks_proto, result = client._serialize(hook_name, extras)
expected_result = msgpack.packb({
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 'method': hook_name,
python3: code change for py3 support...
r1048 'extras': extras,
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 })
tests: test fixes after py311 compat run
r1115 assert hooks_proto == {'rc-hooks-protocol': 'msgpack.v1', 'Connection': 'keep-alive'}
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 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)
python3: code change for py3 support...
r1048 expected_result = msgpack.unpackb(msgpack.packb({
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 'method': hook_name,
'extras': extras
python3: code change for py3 support...
r1048 }), raw=False)
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 assert result == expected_result
lint: auto-fixes
r1152 class TestHooksDummyClient:
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 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):
python3: code change for py3 support...
r1048
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 def do_POST(self):
length = int(self.headers['Content-Length'])
python3: code change for py3 support...
r1048 body = self.rfile.read(length)
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 self.send_response(200)
self.end_headers()
self.wfile.write(body)
lint: auto-fixes
r1152 class MirrorHttpServer:
packaging: moved tests into the main library itself. This is consistent with how our other projects do it.
r130 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)
tests: added few generated tests
r1087
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 \
tests: test fixes after py311 compat run
r1115 mock.patch('http.client.HTTPConnection') as mock_connection,\
tests: added few generated tests
r1087 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)
tests: test fixes after py311 compat run
r1115 assert headers == {'rc-hooks-protocol': HooksHttpClient.proto, 'Connection': 'keep-alive'}
tests: added few generated tests
r1087 assert msgpack.unpackb(body) == {'method': method, 'extras': extras}