Show More
test_repo.py
120 lines
| 4.2 KiB
| text/x-python
|
PythonLexer
r375 | # -*- coding: utf-8 -*- | |||
r1271 | # Copyright (C) 2010-2017 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 | ||||
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, | |||
RepoPreDeleteEvent, RepoDeleteEvent, | ||||
r375 | ) | |||
@pytest.fixture | ||||
def scm_extras(user_regular, repo_stub): | ||||
extras = utils2.AttributeDict({ | ||||
'ip': '127.0.0.1', | ||||
'username': user_regular.username, | ||||
'action': '', | ||||
'repository': repo_stub.repo_name, | ||||
'scm': repo_stub.scm_instance().alias, | ||||
'config': '', | ||||
'server_url': 'http://example.com', | ||||
'make_lock': None, | ||||
r1710 | 'user-agent': 'some-client', | |||
r375 | 'locked_by': [None], | |||
'commit_ids': ['a' * 40] * 3, | ||||
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() | ||||
r411 | assert event_catcher.events_types == [RepoPreCreateEvent, RepoCreateEvent] | |||
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 | ||||