Implemented the observer pattern.

pypatterns/behavioral/observer.py

@@ -0,0 +1,55 @@
+from abc import ABCMeta, abstractmethod
+
+
+class Observer(object, metaclass=ABCMeta):
+    """
+    Abstract Observer class as part of the Observer design pattern.
+    """
+    @abstractmethod
+    def update(self, **state):
+        """
+        Abstract method that is called when an Observable's state changes.
+        """
+        pass
+
+
+class Observable(object):
+    """
+    Base Observable class as part of the Observer design pattern
+    """
+    def __init__(self):
+        """
+        Initialize a new Observable instance.
+        """
+        self._observers = set()
+
+    def attach(self, observer):
+        """
+        Attach an observer to this Observable.
+
+        @param observer: The Observer to attach.
+        @type observer: Observer
+        """
+        self._observers.add(observer)
+
+    def detach(self, observer):
+        """
+        Detach an observer from this Observable.
+
+        @param observer: The Observer to detach.
+        @type observer: Observer
+        """
+        try:
+            self._observers.remove(observer)
+        except KeyError:
+            pass
+
+    def notify(self):
+        """
+        Notify all attached Observers of the state of this Observable.
+        This should be called when this Observable's state changes.
+        """
+        for observer in self._observers:
+            state = {k: v for k, v in self.__dict__.items() if not k.startswith('__') and not k.startswith('_')}
+            observer.update(**state)
+

tests/behavioral_tests/test_observer.py

@@ -0,0 +1,139 @@
+from unittest import TestCase
+from pypatterns.behavioral.observer import Observer, Observable
+
+
+class ObserverTestCase(TestCase):
+    """
+    Unit testing class for the Observer class.
+    """
+    def setUp(self):
+        """
+        Initialize testing data.
+        """
+        class ConcreteObserver(Observer):
+
+            updated_state = None
+
+            def update(self, **state):
+                self.updated_state = state
+
+        self.observer = ConcreteObserver()
+
+    def test_update(self):
+        """
+        Test the update method.
+
+        @raise AssertionError: If the test fails.
+        """
+        state = {'foo': 'test1', 'bar': 'test2'}
+        self.observer.update(**state)
+
+        self.assertEquals(state, self.observer.updated_state)
+
+
+class ObservableTestCase(TestCase):
+    """
+    Unit testing class for the Observable class.
+    """
+    def setUp(self):
+        """
+        Initialize testing data.
+        """
+        class ConcreteObservable(Observable):
+            _kinda_private_var = 'I am kinda private'
+            __private_var = True
+
+            def change_state(self, **kwargs):
+                for key, value in kwargs.items():
+                    setattr(self, key, value)
+
+                self.notify()
+
+        class ConcreteObserver(Observer):
+
+            updated_state = None
+
+            def update(self, **state):
+                self.updated_state = state
+
+        self.observer_class = ConcreteObserver
+        self.observable_class = ConcreteObservable
+
+    def test_attach(self):
+        """
+        Test the attach method.
+
+        @raise AssertionError: If the test fails.
+        """
+        observable = self.observable_class()
+        observer_1 = self.observer_class()
+        observer_2 = self.observer_class()
+        observer_3 = self.observer_class()
+
+        observable.attach(observer_1)
+        observable.attach(observer_2)
+        observable.attach(observer_3)
+
+        try:
+            observable.attach(observer_1)
+        except:
+            raise AssertionError
+        else:
+            self.assertEquals({observer_1, observer_2, observer_3}, observable._observers)
+
+    def test_detach(self):
+        """
+        Test the detach method.
+
+        @raise AssertionError: If the test fails.
+        """
+        observable = self.observable_class()
+        observer_1 = self.observer_class()
+        observer_2 = self.observer_class()
+        observer_3 = self.observer_class()
+        observer_unattached = self.observer_class()
+
+        observable.attach(observer_1)
+        observable.attach(observer_2)
+        observable.attach(observer_3)
+
+        observable.detach(observer_1)
+        observable.detach(observer_2)
+        observable.detach(observer_3)
+
+        try:
+            observable.detach(observer_unattached)
+        except:
+            raise AssertionError
+        else:
+            self.assertEquals(set(), observable._observers)
+
+    def test_notify(self):
+        """
+        Test the notify method.
+
+        @raise AssertionError: If the test fails.
+        """
+        observable = self.observable_class()
+        observer_1 = self.observer_class()
+        observer_2 = self.observer_class()
+        observer_3 = self.observer_class()
+
+        observable.attach(observer_1)
+        observable.attach(observer_2)
+        observable.attach(observer_3)
+
+        observable.change_state(public_state={'foo': 'test1', 'bar': 'test2'}, foo='foo', bar=False)
+        expected_state = {'public_state': {'foo': 'test1', 'bar': 'test2'}, 'foo': 'foo', 'bar': False}
+
+        self.assertDictEqual(expected_state, observer_1.updated_state)
+        self.assertDictEqual(expected_state, observer_2.updated_state)
+        self.assertDictEqual(expected_state, observer_3.updated_state)
+
+        observable.change_state(bar='bar')
+        expected_state_2 = {'public_state': {'foo': 'test1', 'bar': 'test2'}, 'foo': 'foo', 'bar': 'bar'}
+
+        self.assertDictEqual(expected_state_2, observer_1.updated_state)
+        self.assertDictEqual(expected_state_2, observer_2.updated_state)
+        self.assertDictEqual(expected_state_2, observer_3.updated_state)
+