Show More
test_hooks_base.py
151 lines
| 4.9 KiB
| text/x-python
|
PythonLexer
r1 | # -*- coding: utf-8 -*- | |||
r2487 | # Copyright (C) 2010-2018 RhodeCode GmbH | |||
r1 | # | |||
# 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 <http://www.gnu.org/licenses/>. | ||||
# | ||||
# 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 mock | ||||
Martin Bornhold
|
r910 | import pytest | ||
r1754 | from rhodecode.model.db import Session, UserLog | |||
r1 | from rhodecode.lib import hooks_base, utils2 | |||
r375 | def test_post_push_truncates_commits(user_regular, repo_stub): | |||
r1 | extras = { | |||
'ip': '127.0.0.1', | ||||
r375 | 'username': user_regular.username, | |||
r2411 | 'user_id': user_regular.user_id, | |||
r1 | 'action': 'push_local', | |||
r375 | 'repository': repo_stub.repo_name, | |||
r1 | 'scm': 'git', | |||
'config': '', | ||||
'server_url': 'http://example.com', | ||||
'make_lock': None, | ||||
r1710 | 'user_agent': 'some-client', | |||
r1 | 'locked_by': [None], | |||
'commit_ids': ['abcde12345' * 4] * 30000, | ||||
r3133 | 'hook_type': 'large_push_test_type', | |||
Martin Bornhold
|
r909 | 'is_shadow_repo': False, | ||
r1 | } | |||
extras = utils2.AttributeDict(extras) | ||||
hooks_base.post_push(extras) | ||||
# Calculate appropriate action string here | ||||
r1964 | commit_ids = extras.commit_ids[:400] | |||
r1 | ||||
r1754 | entry = UserLog.query().order_by('-user_log_id').first() | |||
assert entry.action == 'user.push' | ||||
assert entry.action_data['commit_ids'] == commit_ids | ||||
Session().delete(entry) | ||||
Session().commit() | ||||
Martin Bornhold
|
r910 | |||
def assert_called_with_mock(callable_, expected_mock_name): | ||||
mock_obj = callable_.call_args[0][0] | ||||
mock_name = mock_obj._mock_new_parent._mock_new_name | ||||
assert mock_name == expected_mock_name | ||||
@pytest.fixture | ||||
def hook_extras(user_regular, repo_stub): | ||||
extras = utils2.AttributeDict({ | ||||
'ip': '127.0.0.1', | ||||
'username': user_regular.username, | ||||
r2411 | 'user_id': user_regular.user_id, | |||
Martin Bornhold
|
r910 | 'action': 'push', | ||
'repository': repo_stub.repo_name, | ||||
'scm': '', | ||||
'config': '', | ||||
r3094 | 'repo_store': '', | |||
Martin Bornhold
|
r910 | 'server_url': 'http://example.com', | ||
'make_lock': None, | ||||
r1710 | 'user_agent': 'some-client', | |||
Martin Bornhold
|
r910 | 'locked_by': [None], | ||
'commit_ids': [], | ||||
r3133 | 'hook_type': 'test_type', | |||
Martin Bornhold
|
r910 | 'is_shadow_repo': False, | ||
}) | ||||
return extras | ||||
@pytest.mark.parametrize('func, extension, event', [ | ||||
(hooks_base.pre_push, 'pre_push_extension', 'RepoPrePushEvent'), | ||||
(hooks_base.post_push, 'post_pull_extension', 'RepoPushEvent'), | ||||
(hooks_base.pre_pull, 'pre_pull_extension', 'RepoPrePullEvent'), | ||||
(hooks_base.post_pull, 'post_push_extension', 'RepoPullEvent'), | ||||
]) | ||||
def test_hooks_propagate(func, extension, event, hook_extras): | ||||
""" | ||||
Tests that our hook code propagates to rhodecode extensions and triggers | ||||
the appropriate event. | ||||
""" | ||||
r3133 | class ExtensionMock(mock.Mock): | |||
@property | ||||
def output(self): | ||||
return 'MOCK' | ||||
extension_mock = ExtensionMock() | ||||
Martin Bornhold
|
r910 | events_mock = mock.Mock() | ||
patches = { | ||||
'Repository': mock.Mock(), | ||||
'events': events_mock, | ||||
extension: extension_mock, | ||||
} | ||||
# Clear shadow repo flag. | ||||
hook_extras.is_shadow_repo = False | ||||
# Execute hook function. | ||||
with mock.patch.multiple(hooks_base, **patches): | ||||
func(hook_extras) | ||||
# Assert that extensions are called and event was fired. | ||||
extension_mock.called_once() | ||||
assert_called_with_mock(events_mock.trigger, event) | ||||
@pytest.mark.parametrize('func, extension, event', [ | ||||
(hooks_base.pre_push, 'pre_push_extension', 'RepoPrePushEvent'), | ||||
(hooks_base.post_push, 'post_pull_extension', 'RepoPushEvent'), | ||||
(hooks_base.pre_pull, 'pre_pull_extension', 'RepoPrePullEvent'), | ||||
(hooks_base.post_pull, 'post_push_extension', 'RepoPullEvent'), | ||||
]) | ||||
def test_hooks_propagates_not_on_shadow(func, extension, event, hook_extras): | ||||
""" | ||||
If hooks are called by a request to a shadow repo we only want to run our | ||||
internal hooks code but not external ones like rhodecode extensions or | ||||
trigger an event. | ||||
""" | ||||
extension_mock = mock.Mock() | ||||
events_mock = mock.Mock() | ||||
patches = { | ||||
'Repository': mock.Mock(), | ||||
'events': events_mock, | ||||
extension: extension_mock, | ||||
} | ||||
# Set shadow repo flag. | ||||
hook_extras.is_shadow_repo = True | ||||
# Execute hook function. | ||||
with mock.patch.multiple(hooks_base, **patches): | ||||
func(hook_extras) | ||||
# Assert that extensions are *not* called and event was *not* fired. | ||||
assert not extension_mock.called | ||||
assert not events_mock.trigger.called | ||||