Skip to content

Commit

Permalink
feat: add event bus backend
Browse files Browse the repository at this point in the history
  • Loading branch information
Ian2012 committed Oct 25, 2023
1 parent 6d107ec commit 3d31a21
Show file tree
Hide file tree
Showing 2 changed files with 69 additions and 0 deletions.
28 changes: 28 additions & 0 deletions eventtracking/backends/event_bus.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
"""Event tracker backend that emits events to the event-bus."""
from openedx_events.analytics.signals import TRACKING_EVENT_EMITTED
from openedx_events.analytics.data import TrackingLogData

class EventBusRoutingBackend:
"""
Event tracker backend that emits an Open edX public signal.
"""

def __init__(self, **kwargs):
"""
Event tracker backend that emits an Open edX public signal.
"""

def send(self, event):
"""
Emit the TRACKING_EVENT_EMITTED Open edX public signal to allow
other apps to listen for tracking events.
"""
# .. event_implemented_name: TRACKING_EVENT_EMITTED
TRACKING_EVENT_EMITTED.send_event(
tracking_log=TrackingLogData(
name=event.get('name'),
timestamp=event.get('timestamp'),
data=event.get('data'),
context=event.get('context')
)
)
41 changes: 41 additions & 0 deletions eventtracking/backends/tests/test_event_bus.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
"""
Test the async routing backend.
"""
from unittest import TestCase

from unittest.mock import sentinel, patch
from eventtracking.backends.event_bus import EventBusRoutingBackend
from openedx_events.analytics.data import TrackingLogData

class TestAsyncRoutingBackend(TestCase):
"""
Test the async routing backend.
"""

def setUp(self):
super().setUp()
self.sample_event = {
'name': str(sentinel.name),
'data': 'data',
'timestamp': '2020-01-01T12:12:12.000000+00:00',
'context': {},
}

@patch('eventtracking.backends.event_bus.EventBusRoutingBackend.send')
def test_successful_send(self, mocked_send_event):
backend = EventBusRoutingBackend()
backend.send(self.sample_event)
mocked_send_event.assert_called_once_with(self.sample_event)

@patch('eventtracking.backends.event_bus.TRACKING_EVENT_EMITTED.send_event')
def test_successful_send_event(self, mocked_send_event):
backend = EventBusRoutingBackend()
backend.send(self.sample_event)
mocked_send_event.assert_called_once_with(
tracking_log=TrackingLogData(
name=self.sample_event['name'],
timestamp=self.sample_event['timestamp'],
data=self.sample_event['data'],
context=self.sample_event['context']
)
)

0 comments on commit 3d31a21

Please sign in to comment.