10. Analytics¶

def assert_event_matches(expected_event, actual_event, tolerate=None):

# By default, decode string values for the "event" field as JSON and compare
# the contents with the actual event. This will not raise an error.
assert_event_matches(
    {'event': {'a': 'b'}},
    {'event': '{"a": "b"}'}
)

# Ignore "unexpected" root fields. This will not raise an error even though
# the field "foo" does not appear in the expected event.
assert_event_matches(
    {'event_type': 'test'},
    {'event_type': 'test', 'foo': 'bar'}
)

# Ignore "unexpected" fields in the context. This will not raise an error
# even though the field "foo" does not appear in the expected event context.
assert_event_matches(
    {'event_type': 'test'},
    {'event_type': 'test', 'context': {'foo': 'bar'}}
)

# Overriding "tolerate" allows more strict assertions to be made.
# This assertion will raise an error!
assert_event_matches(
    {'event_type': 'test'},
    {'event_type': 'test', 'context': {'foo': 'bar'}},
    tolerate=[]
)

from track.tests import EventTrackingTestCase
from openedx.core.lib.tests.assertions.events import assert_event_matches

class MyTestClass(EventTrackingTestCase):

    def setUp(self):
        # The setUp() of the superclass must be called
        super(MyTestClass, self).setUp()

    def test_event_emitted(self):
        my_function_that_emits_events()

        # If the above function only emits a single event, this can be used.
        actual_event = self.get_event()

        # This will assert that the "event_type" of the event is "foobar".
        # Note that it makes no assertions about any of the other fields
        # in the event.
        assert_event_matches({'event_type': 'foobar'}, actual_event)

    def test_no_event_emitted(self):

        my_function_that_does_not_emit()

        # This will fail if any events were emitted by the above function
        # call.
        self.assert_no_events_emitted()

def test_foobar_event_emission(self):
    emit_foobar_event()

    # This will wait for the event to be emitted. It will time out if the
    # event is not emitted quickly enough (or not emitted at all).
    actual_events = self.wait_for_events({'event_type': 'foobar'})

    # This will compare the first event emitted with the first expected
    # event, the second with the second etc.
    self.assert_events_match(
      [
        {'event': {'a': 'b'}}
      ],
      actual_events
    )

    # ``wait_for_events`` also accepts arbitrary callable functions to check
    # to see if an event "matches"
    def some_custom_event_filter(event):
        return event['event']['old_time'] > 10

    # This will return when some_custom_event_filter returns true for at
    # at least one event.
    actual_events = self.wait_for_events(some_custom_event_filter)

def test_multiple_events(self):
    emit_several_events()

    def my_event_filter(event):
        return event['event_type'] in ('first_event', 'second_event')

    # This will wait for 2 events to match the filter defined above. Note
    # that it makes no assertions about their ordering or content.
    actual_events = self.wait_for_events(my_event_filter, number_of_matches=2)

    # This ensures that first_event was emitted before second_event and
    # checks the payload of both events.
    self.assert_events_match(
      [
        {
          'event_type': 'first_event',
          'event': {'a': 'b'}
        },
        {
          'event_type': 'second_event',
          'event': {'a': 'other'}
        }
      ],
      actual_events
    )

def test_granular_assertion(self):

    # This foobar event is emitted first, with the "a" field set to "NOT B"
    tracker.emit('foobar', {'a': 'NOT B'})

    # A context manager can be used to ensure that the first "foobar" event
    # is ignored. It only makes assertions about the events that are emitted
    # inside this context.
    with self.assert_events_match_during(
        {'event_type': 'foobar'},
        [
          {
            'event': {'a': 'b'}
          }
        ]
    ):
        emit_foobar_event()