Show More
test_repo.py
170 lines
| 5.7 KiB
| text/x-python
|
PythonLexer
r375 | ||||
r5088 | # Copyright (C) 2010-2023 RhodeCode GmbH | |||
r375 | # | |||
# 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 pytest | ||||
r4704 | from rhodecode.events import UserPermissionsChange | |||
r4314 | from rhodecode.lib.utils2 import StrictAttributeDict | |||
r376 | from rhodecode.tests.events.conftest import EventCatcher | |||
r375 | ||||
from rhodecode.lib import hooks_base, utils2 | ||||
from rhodecode.model.repo import RepoModel | ||||
from rhodecode.events.repo import ( | ||||
RepoPrePullEvent, RepoPullEvent, | ||||
RepoPrePushEvent, RepoPushEvent, | ||||
r411 | RepoPreCreateEvent, RepoCreateEvent, | |||
r4444 | RepoPreDeleteEvent, RepoDeleteEvent, | |||
RepoCommitCommentEvent, RepoCommitCommentEditEvent | ||||
r375 | ) | |||
r3946 | @pytest.fixture() | |||
r375 | def scm_extras(user_regular, repo_stub): | |||
extras = utils2.AttributeDict({ | ||||
'ip': '127.0.0.1', | ||||
'username': user_regular.username, | ||||
r2411 | 'user_id': user_regular.user_id, | |||
r375 | 'action': '', | |||
'repository': repo_stub.repo_name, | ||||
'scm': repo_stub.scm_instance().alias, | ||||
'config': '', | ||||
r3094 | 'repo_store': '', | |||
r375 | 'server_url': 'http://example.com', | |||
'make_lock': None, | ||||
r3133 | 'user_agent': 'some-client', | |||
r375 | 'locked_by': [None], | |||
'commit_ids': ['a' * 40] * 3, | ||||
r3133 | 'hook_type': 'scm_extras_test', | |||
Martin Bornhold
|
r909 | 'is_shadow_repo': False, | ||
r375 | }) | |||
return extras | ||||
r389 | # TODO: dan: make the serialization tests complete json comparisons | |||
@pytest.mark.parametrize('EventClass', [ | ||||
r411 | RepoPreCreateEvent, RepoCreateEvent, | |||
RepoPreDeleteEvent, RepoDeleteEvent, | ||||
r389 | ]) | |||
r1785 | def test_repo_events_serialized(config_stub, repo_stub, EventClass): | |||
r389 | event = EventClass(repo_stub) | |||
data = event.as_dict() | ||||
assert data['name'] == EventClass.name | ||||
assert data['repo']['repo_name'] == repo_stub.repo_name | ||||
r392 | assert data['repo']['url'] | |||
r1788 | assert data['repo']['permalink_url'] | |||
r389 | ||||
@pytest.mark.parametrize('EventClass', [ | ||||
RepoPrePullEvent, RepoPullEvent, RepoPrePushEvent | ||||
]) | ||||
r1785 | def test_vcs_repo_events_serialize(config_stub, repo_stub, scm_extras, EventClass): | |||
r389 | event = EventClass(repo_name=repo_stub.repo_name, extras=scm_extras) | |||
data = event.as_dict() | ||||
assert data['name'] == EventClass.name | ||||
assert data['repo']['repo_name'] == repo_stub.repo_name | ||||
r392 | assert data['repo']['url'] | |||
r1788 | assert data['repo']['permalink_url'] | |||
r389 | ||||
@pytest.mark.parametrize('EventClass', [RepoPushEvent]) | ||||
r1785 | def test_vcs_repo_push_event_serialize(config_stub, repo_stub, scm_extras, EventClass): | |||
r389 | event = EventClass(repo_name=repo_stub.repo_name, | |||
pushed_commit_ids=scm_extras['commit_ids'], | ||||
extras=scm_extras) | ||||
data = event.as_dict() | ||||
assert data['name'] == EventClass.name | ||||
assert data['repo']['repo_name'] == repo_stub.repo_name | ||||
r392 | assert data['repo']['url'] | |||
r1788 | assert data['repo']['permalink_url'] | |||
r389 | ||||
r375 | def test_create_delete_repo_fires_events(backend): | |||
r376 | with EventCatcher() as event_catcher: | |||
repo = backend.create_repo() | ||||
r4704 | assert event_catcher.events_types == [RepoPreCreateEvent, RepoCreateEvent, UserPermissionsChange] | |||
r376 | ||||
with EventCatcher() as event_catcher: | ||||
RepoModel().delete(repo) | ||||
r411 | assert event_catcher.events_types == [RepoPreDeleteEvent, RepoDeleteEvent] | |||
r375 | ||||
def test_pull_fires_events(scm_extras): | ||||
r376 | with EventCatcher() as event_catcher: | |||
hooks_base.pre_push(scm_extras) | ||||
r377 | assert event_catcher.events_types == [RepoPrePushEvent] | |||
r376 | ||||
with EventCatcher() as event_catcher: | ||||
hooks_base.post_push(scm_extras) | ||||
r377 | assert event_catcher.events_types == [RepoPushEvent] | |||
r375 | ||||
def test_push_fires_events(scm_extras): | ||||
r376 | with EventCatcher() as event_catcher: | |||
hooks_base.pre_pull(scm_extras) | ||||
r377 | assert event_catcher.events_types == [RepoPrePullEvent] | |||
r376 | ||||
with EventCatcher() as event_catcher: | ||||
hooks_base.post_pull(scm_extras) | ||||
r377 | assert event_catcher.events_types == [RepoPullEvent] | |||
r389 | ||||
r4314 | ||||
@pytest.mark.parametrize('EventClass', [RepoCommitCommentEvent]) | ||||
def test_repo_commit_event(config_stub, repo_stub, EventClass): | ||||
commit = StrictAttributeDict({ | ||||
'raw_id': 'raw_id', | ||||
'message': 'message', | ||||
'branch': 'branch', | ||||
}) | ||||
comment = StrictAttributeDict({ | ||||
'comment_id': 'comment_id', | ||||
'text': 'text', | ||||
'comment_type': 'comment_type', | ||||
'f_path': 'f_path', | ||||
'line_no': 'line_no', | ||||
r4444 | 'last_version': 0, | |||
r4314 | }) | |||
event = EventClass(repo=repo_stub, commit=commit, comment=comment) | ||||
data = event.as_dict() | ||||
assert data['commit']['commit_id'] | ||||
assert data['comment']['comment_id'] | ||||
r4444 | ||||
@pytest.mark.parametrize('EventClass', [RepoCommitCommentEditEvent]) | ||||
def test_repo_commit_edit_event(config_stub, repo_stub, EventClass): | ||||
commit = StrictAttributeDict({ | ||||
'raw_id': 'raw_id', | ||||
'message': 'message', | ||||
'branch': 'branch', | ||||
}) | ||||
comment = StrictAttributeDict({ | ||||
'comment_id': 'comment_id', | ||||
'text': 'text', | ||||
'comment_type': 'comment_type', | ||||
'f_path': 'f_path', | ||||
'line_no': 'line_no', | ||||
'last_version': 0, | ||||
}) | ||||
event = EventClass(repo=repo_stub, commit=commit, comment=comment) | ||||
data = event.as_dict() | ||||
assert data['commit']['commit_id'] | ||||
assert data['comment']['comment_id'] | ||||