events: use context manager for testing fired events

rhodecode/events/__init__.py

@@ -1,49 +1,51 b''
 # Copyright (C) 2016-2016  RhodeCode GmbH
 #
 # 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/
 
 from pyramid.threadlocal import get_current_registry
 
 
 class RhodecodeEvent(object):
     """
     Base event class for all Rhodecode events
     """
 
 
 def trigger(event):
     """
     Helper method to send an event. This wraps the pyramid logic to send an
     event.
     """
     # For the first step we are using pyramids thread locals here. If the
     # event mechanism works out as a good solution we should think about
     # passing the registry as an argument to get rid of it.
     registry = get_current_registry()
     registry.notify(event)
 
 
 from rhodecode.events.user import (
-    UserPreCreate, UserPreUpdate, UserRegistered
+    UserPreCreate,
+    UserPreUpdate,
+    UserRegistered
 )
 
 from rhodecode.events.repo import (
     RepoPreCreateEvent, RepoCreatedEvent,
     RepoPreDeleteEvent, RepoDeletedEvent,
     RepoPrePushEvent,   RepoPushEvent,
     RepoPrePullEvent,   RepoPullEvent,
-)
  No newline at end of file
+)

rhodecode/events/repo.py

@@ -1,115 +1,113 b''
 # Copyright (C) 2016-2016  RhodeCode GmbH
 #
 # 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/
 
 from rhodecode.model.db import Repository, Session
 from rhodecode.events import RhodecodeEvent
 
 
 class RepoEvent(RhodecodeEvent):
     """
     Base class for events acting on a repository.
 
     :param repo: a :class:`Repository` instance
     """
     def __init__(self, repo):
         self.repo = repo
 
 
 class RepoPreCreateEvent(RepoEvent):
     """
     An instance of this class is emitted as an :term:`event` before a repo is
     created.
-
-    :param repo_name: repository name
     """
     name = 'repo-pre-create'
 
 
 class RepoCreatedEvent(RepoEvent):
     """
     An instance of this class is emitted as an :term:`event` whenever a repo is
     created.
     """
     name = 'repo-created'
 
 
 class RepoPreDeleteEvent(RepoEvent):
     """
     An instance of this class is emitted as an :term:`event` whenever a repo is
     created.
     """
     name = 'repo-pre-delete'
 
 
 class RepoDeletedEvent(RepoEvent):
     """
     An instance of this class is emitted as an :term:`event` whenever a repo is
     created.
     """
     name = 'repo-deleted'
 
 
 class RepoVCSEvent(RepoEvent):
     """
     Base class for events triggered by the VCS
     """
     def __init__(self, repo_name, extras):
         self.repo = Repository.get_by_repo_name(repo_name)
         if not self.repo:
             raise Exception('repo by this name %s does not exist' % repo_name)
         self.extras = extras
         super(RepoVCSEvent, self).__init__(self.repo)
 
 
 class RepoPrePullEvent(RepoVCSEvent):
     """
     An instance of this class is emitted as an :term:`event` before commits
     are pulled from a repo.
     """
     name = 'repo-pre-pull'
 
 
 class RepoPullEvent(RepoVCSEvent):
     """
     An instance of this class is emitted as an :term:`event` after commits
     are pulled from a repo.
     """
     name = 'repo-pull'
 
 
 class RepoPrePushEvent(RepoVCSEvent):
     """
     An instance of this class is emitted as an :term:`event` before commits
     are pushed to a repo.
     """
     name = 'repo-pre-push'
 
 
 class RepoPushEvent(RepoVCSEvent):
     """
     An instance of this class is emitted as an :term:`event` after commits
     are pushed to a repo.
 
     :param extras: (optional) dict of data from proxied VCS actions
     """
     name = 'repo-push'
 
     def __init__(self, repo_name, pushed_commit_ids, extras):
         super(RepoPushEvent, self).__init__(repo_name, extras)
         self.pushed_commit_ids = pushed_commit_ids
 

rhodecode/tests/events/conftest.py

@@ -1,38 +1,41 b''
 # Copyright (C) 2016-2016  RhodeCode GmbH
 #
 # 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
 import decorator
 
 
-def assert_fires_events(*expected_events):
-    """ Testing decorator to check if the function fires events in order """
-    def deco(func):
-        def wrapper(func, *args, **kwargs):
-            with mock.patch('rhodecode.events.trigger') as mock_trigger:
-                result = func(*args, **kwargs)
+class EventCatcher(object):
+    """ Testing context manager to check if events are fired """
+
+    def __init__(self):
+        self.events = [] # the actual events captured
+        self.event_types = [] # the types of events captured
 
-            captured_events = []
-            for call in mock_trigger.call_args_list:
-                event = call[0][0]
-                captured_events.append(type(event))
+    def __enter__(self):
+        self.event_trigger_patch = mock.patch('rhodecode.events.trigger')
+        self.mocked_event_trigger = self.event_trigger_patch.start()
+        return self
 
-            assert set(captured_events) == set(expected_events)
-            return result
-        return decorator.decorator(wrapper, func)
-    return deco
  No newline at end of file
+    def __exit__(self, type_, value, traceback):
+        self.event_trigger_patch.stop()
+
+        for call in self.mocked_event_trigger.call_args_list:
+            event = call[0][0]
+            self.events.append(event)
+            self.event_types.append(type(event))

rhodecode/tests/events/test_repo.py

@@ -1,68 +1,79 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  RhodeCode GmbH
 #
 # 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
 
-from rhodecode.tests.events.conftest import assert_fires_events
+from rhodecode.tests.events.conftest import EventCatcher
 
 from rhodecode.lib import hooks_base, utils2
 from rhodecode.model.repo import RepoModel
 from rhodecode.events.repo import (
     RepoPrePullEvent, RepoPullEvent,
     RepoPrePushEvent, RepoPushEvent,
     RepoPreCreateEvent, RepoCreatedEvent,
     RepoPreDeleteEvent, RepoDeletedEvent,
 )
 
 
 @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,
         'locked_by': [None],
         'commit_ids': ['a' * 40] * 3,
     })
     return extras
 
 
-@assert_fires_events(
-    RepoPreCreateEvent, RepoCreatedEvent, RepoPreDeleteEvent, RepoDeletedEvent)
 def test_create_delete_repo_fires_events(backend):
+    with EventCatcher() as event_catcher:
     repo = backend.create_repo()
+    assert event_catcher.event_types == [RepoPreCreateEvent, RepoCreatedEvent]
+
+    with EventCatcher() as event_catcher:
     RepoModel().delete(repo)
+    assert event_catcher.event_types == [RepoPreDeleteEvent, RepoDeletedEvent]
 
 
-@assert_fires_events(RepoPrePushEvent, RepoPushEvent)
 def test_pull_fires_events(scm_extras):
+    with EventCatcher() as event_catcher:
     hooks_base.pre_push(scm_extras)
+    assert event_catcher.event_types == [RepoPrePushEvent]
+
+    with EventCatcher() as event_catcher:
     hooks_base.post_push(scm_extras)
+    assert event_catcher.event_types == [RepoPushEvent]
 
 
-@assert_fires_events(RepoPrePullEvent, RepoPullEvent)
 def test_push_fires_events(scm_extras):
+    with EventCatcher() as event_catcher:
     hooks_base.pre_pull(scm_extras)
+    assert event_catcher.event_types == [RepoPrePullEvent]
+
+    with EventCatcher() as event_catcher:
     hooks_base.post_pull(scm_extras)
+    assert event_catcher.event_types == [RepoPullEvent]