Observer design pattern in Python.

observable.py

#!/usr/bin/env python2.6
"""observable.py An implementation of the Observer design pattern in
Python.

"""
class Observable:
    """Implements the observer design pattern: maps events (strings) to
    sets of callback functions. When an event is fired each callback
    function subscribed to that event is called.

    User classes can wrap an Observable object, forward the subscribe()
    and unsubscribe() methods and use the fire() method.

    """
    def __init__(self):
        self._callbacks = {}

    def subscribe(self,event,callback):
        """Subscribe a callback function to an event. The function will
        be called whenever the event is fired.

        """
        if event not in self._callbacks:
            self._callbacks[event] = set([callback])
        else:
            self._callbacks[event].add(callback)

    def unsubscribe(self,event,callback):
        """Unsubscribe a callback function from an event. The function
        will no longer be called when the event is fired.

        Raises KeyError if the given callback is not subscribed to the
        given event.

        """
        self._callbacks[event].remove(callback)

    def fire(self,event,*args):
        """Fire an event: call each of the callback functions subscribed
        to the given event, passing *args to each.

        Raises KeyError if there are no callbacks subscribed to the
        given event.

        """
        for callback in self._callbacks[event]:
            callback(*args)

observable_test.py

#!/usr/bin/env python2.6
"""observable_test.py Unit tests for observable.py.

"""
import observable
import unittest

class ObservableTest(unittest.TestCase):

    def callback_1(self,*args):
        if len(args) > 0:
            self.calls['callback_1'].append(*args)
        else:
            self.calls['callback_1'].append(None)

    def callback_2(self,*args):
        if len(args) > 0:
            self.calls['callback_2'].append(*args)
        else:
            self.calls['callback_2'].append(None)

    def callback_3(self,*args):
        if len(args) > 0:
            self.calls['callback_3'].append(*args)
        else:
            self.calls['callback_3'].append(None)

    def setUp(self):
        self.observable = observable.Observable()
        self.calls = {'callback_1':[],'callback_2':[],'callback_3':[]}
        self.observable.subscribe('event_one',self.callback_1)
        self.observable.subscribe('event_one',self.callback_2)
        self.observable.subscribe('event_one',self.callback_3)
        self.observable.subscribe('event_two',self.callback_1)
        self.observable.subscribe('event_two',self.callback_2)
        self.observable.subscribe('event_three',self.callback_3)

    def tearDown(self):
        pass

    def testFire(self):
        self.observable.fire('event_one','fsdgs')
        self.assertEqual(self.calls['callback_1'], ['fsdgs'])
        self.assertEqual(self.calls['callback_2'], ['fsdgs'])
        self.assertEqual(self.calls['callback_3'], ['fsdgs'])
        self.observable.fire('event_two','dddd')
        self.assertEqual(self.calls['callback_1'], ['fsdgs','dddd'])
        self.assertEqual(self.calls['callback_2'], ['fsdgs','dddd'])
        self.assertEqual(self.calls['callback_3'], ['fsdgs'])
        self.observable.fire('event_three','xxxxx')
        self.assertEqual(self.calls['callback_1'], ['fsdgs','dddd'])
        self.assertEqual(self.calls['callback_2'], ['fsdgs','dddd'])
        self.assertEqual(self.calls['callback_3'], ['fsdgs','xxxxx'])

    def testUnsubscribe(self):
        self.observable.unsubscribe('event_one',self.callback_1)
        self.observable.unsubscribe('event_one',self.callback_3)
        self.observable.unsubscribe('event_two',self.callback_2)

        self.observable.fire('event_one','fsdgs')
        self.assertEqual(self.calls['callback_1'], [])
        self.assertEqual(self.calls['callback_2'], ['fsdgs'])
        self.assertEqual(self.calls['callback_3'], [])
        self.observable.fire('event_two','dddd')
        self.assertEqual(self.calls['callback_1'], ['dddd'])
        self.assertEqual(self.calls['callback_2'], ['fsdgs'])
        self.assertEqual(self.calls['callback_3'], [])
        self.observable.fire('event_three','xxxxx')
        self.assertEqual(self.calls['callback_1'],['dddd'])
        self.assertEqual(self.calls['callback_2'],['fsdgs'])
        self.assertEqual(self.calls['callback_3'],['xxxxx'])

    def testUnsubscribeBadEvent(self):
        self.assertRaises(KeyError,self.observable.unsubscribe,'bad_event',self.callback_1)

    def testUnsubscribeBadCallback(self):
        self.assertRaises(KeyError,self.observable.unsubscribe,'event_three',self.callback_2)

    def testFireWithNoArgs(self):
        self.observable.fire('event_three')
        self.assertEqual(self.calls['callback_3'],[None])

    def testFireBadEvent(self):
        self.assertRaises(KeyError,self.observable.fire,'bad_event')

if __name__ == "__main__":
    unittest.main()