feat: add signal isolation class and methods

Author: mariajgrimaldi

openedx_events/tests/__init__.py

@@ -0,0 +1,3 @@
+"""
+Test package for openedx-events implementations.
+"""

openedx_events/tests/test_tooling.py

@@ -185,3 +185,18 @@ def test_send_robust_event_with_django(self):
 
         with self.assertWarns(Warning, msg=message):
             self.public_signal.send_robust(sender=Mock())
+
+    @patch("openedx_events.tooling.Signal.send")
+    def test_send_event_disabled(self, send_mock):
+        """
+        This method tests sending an event that has been disabled.
+
+        Expected behavior:
+            The Django Signal associated to the event is not sent.
+        """
+        self.public_signal.disable()
+
+        result = self.public_signal.send_event(sender=Mock())
+
+        send_mock.assert_not_called()
+        self.assertListEqual([], result)

openedx_events/tests/utils.py

@@ -0,0 +1,78 @@
+"""
+Utils used by Open edX events.
+"""
+from django.test import TestCase
+
+from openedx_events.tooling import OpenEdxPublicSignal
+
+
+class EventsIsolationMixin:
+    """
+    A mixin to be used by TestCases that want to isolate their use of Open edX Events.
+    """
+
+    @classmethod
+    def disable_all_events(cls):
+        """
+        Disable all events Open edX Events from all subdomains.
+        """
+        for event in OpenEdxPublicSignal.all_events():
+            event.disable()
+
+    @classmethod
+    def enable_all_events(cls):
+        """
+        Enable all events Open edX Events from all subdomains.
+        """
+        for event in OpenEdxPublicSignal.all_events():
+            event.enable()
+
+    @classmethod
+    def enable_events_by_type(cls, *event_types):
+        """
+        Enable specific Open edX Events given their type.
+
+        Arguments:
+            event_types (list of `str`): types of events to enable.
+        """
+        for event_type in event_types:
+            try:
+                event = OpenEdxPublicSignal.get_signal_by_type(event_type)
+            except KeyError:
+                all_event_types = sorted(s.event_type for s in OpenEdxPublicSignal.all_events())
+                err_msg = (
+                    "You tried to enable event '{}', but I don't recognize that "
+                    "signal type. Did you mean one of these?: {}"
+                )
+                raise ValueError(err_msg.format(event_type, all_event_types))  # pylint: disable=raise-missing-from
+            event.enable()
+
+
+class OpenEdxEventsTestCase(EventsIsolationMixin, TestCase):
+    """
+    A mixin to be used by TestCases that want to isolate their use of Open edX Events.
+
+    Example usage:
+
+        class MyTestCase(OpenEdxEventsTestCase):
+
+            ENABLED_OPENEDX_EVENTS = ['org.openedx.learning.student.registration.completed.v1']
+    """
+
+    ENABLED_OPENEDX_EVENTS = []
+
+    @classmethod
+    def setUpClass(cls):
+        """
+        Start events isolation for class.
+        """
+        super().setUpClass()
+        cls().start_events_isolation()
+
+    @classmethod
+    def start_events_isolation(cls):
+        """
+        Start Open edX Events isolation and then enable events by type.
+        """
+        cls().disable_all_events()
+        cls().enable_events_by_type(*cls.ENABLED_OPENEDX_EVENTS)

openedx_events/tooling.py

@@ -14,6 +14,9 @@ class OpenEdxPublicSignal(Signal):
     Custom class used to create Open edX events.
     """
 
+    _mapping = {}
+    instances = []
+
     def __init__(self, event_type, data, minor_version=0):
         """
         Init method for OpenEdxPublicSignal definition class.
@@ -34,6 +37,9 @@ def __init__(self, event_type, data, minor_version=0):
         self.init_data = data
         self.event_type = event_type
         self.minor_version = minor_version
+        self._allow_events = True
+        self.__class__.instances.append(self)
+        self.__class__._mapping[self.event_type] = self
         super().__init__()
 
     def __repr__(self):
@@ -42,6 +48,20 @@ def __repr__(self):
         """
         return "<OpenEdxPublicSignal: {event_type}>".format(event_type=self.event_type)
 
+    @classmethod
+    def all_events(cls):
+        """
+        Get all current events.
+        """
+        return cls.instances
+
+    @classmethod
+    def get_signal_by_type(cls, event_type):
+        """
+        Get event identified by type.
+        """
+        return cls._mapping[event_type]
+
     def generate_signal_metadata(self):
         """
         Generate signal metadata when an event is sent.
@@ -76,6 +96,8 @@ def send_event(self, send_robust=False, **kwargs):
         some validations are run on the arguments, and then relevant metadata
         that can be used for logging or debugging purposes is generated.
         Besides this behavior, send_event behaves just like the send method.
+        If the event is disabled (i.e _allow_events is False), then this method
+        won't have any effect. Meaning, the Django Signal won't be sent.
 
         Example usage:
             >>> STUDENT_REGISTRATION_COMPLETED.send_event(
@@ -89,7 +111,7 @@ def send_event(self, send_robust=False, **kwargs):
 
         Returns:
             list: response of each receiver following the format
-            [(receiver, response), ... ]
+            [(receiver, response), ... ]. Empty list if the event is disabled.
 
         Exceptions raised:
             SenderValidationError: raised when there's a mismatch between
@@ -126,6 +148,9 @@ def validate_sender():
                         ),
                     )
 
+        if not self._allow_events:
+            return []
+
         validate_sender()
 
         kwargs["metadata"] = self.generate_signal_metadata()
@@ -147,3 +172,15 @@ def send_robust(self, sender, **kwargs):  # pylint: disable=unused-argument
         warnings.warn(
             "Please, use 'send_event' with send_robust equals to True when triggering an Open edX event."
         )
+
+    def enable(self):
+        """
+        Enable all events. Meaning, send_event will send a Django signal.
+        """
+        self._allow_events = True
+
+    def disable(self):
+        """
+        Disable all events. Meaning, send_event will have no effect.
+        """
+        self._allow_events = False