Simplistic Domain Events Python

domain_events_example.py

# coding: utf-8
import unittest
# Unit tests for kata 'domevent'


''' Synchronous domain events in pure Python.
#- triggering an event
#- registering a callback
#- triggering an event with 1 handler
#- triggering an event with 2 handlers
- triggering an event with single argument
#- registering two callbacks for two different events, and triggering one event
- unregistering a callback
'''


# worries
#- static class instead of instance?
#- terminology. want raise, not trigger. is observer accurate enough?


class DomainEvent:
    def __init__(self):
        self.observers = {}

    def add_event_observer(self, event_type, observer):
        if event_type not in self.observers:
            self.observers[event_type] = []
        self.observers[event_type].append(observer)

    def trigger(self, event):
        event_type = type(event)
        if event_type in self.observers:
            for observer in self.observers[event_type]:
                observer(event)

class TestEvent(object):
   pass

class OtherEvent(object):
   pass

class DomeventTests(unittest.TestCase):

    def test_trigger_event(self):
        events = DomainEvent()
        events.trigger(TestEvent())

    def test_add_an_event_observer(self):
        def empty(): pass
        DomainEvent().add_event_observer(TestEvent, empty)

    def test_observe_and_trigger(self):
        called = []
        def observer(evt):
            called.append(True)
        events = DomainEvent()
        events.add_event_observer(TestEvent, observer)
        events.trigger(TestEvent())
        self.assertEqual([True], called)

    def test_observers_and_trigger(self):
        called = []
        def observer1(evt):
            called.append(True)
        def observer2(evt):
            called.append(True)
        events = DomainEvent()
        events.add_event_observer(TestEvent, observer1)
        events.add_event_observer(TestEvent, observer2)
        events.trigger(TestEvent())
        self.assertEqual([True, True], called)


    def test_observering_another_event_does_not_trigger_handling(self):
        called = []
        def observer(evt):
            called.append(True)
        events = DomainEvent()
        events.add_event_observer(TestEvent, observer)
        events.trigger(OtherEvent())
        self.assertEqual([], called)