implemented mediator design pattern

pypatterns/behavioral/mediator.py

@@ -0,0 +1,48 @@
+from collections import defaultdict
+
+
+class Mediator(object):
+    """
+    Mediator class as part of the Mediator design pattern.
+    """
+    def __init__(self):
+        """
+        Initialize a new Mediator instance.
+        """
+        self.signals = defaultdict(list)
+
+    def signal(self, signal_name, *args, **kwargs):
+        """
+        Send a signal out to all connected handlers.
+
+        @param signal_name: The name of the signal.
+        @type signal_name: Str
+        @param args: Positional arguments to send with the signal.
+        @param kwargs: Keyword arguments to send with the signal.
+        """
+        for handler in self.signals[signal_name]:
+            handler(*args, **kwargs)
+
+    def connect(self, signal_name, receiver):
+        """
+        Connect a receiver to a signal.
+
+        @param signal_name: The name of the signal to connect the receiver to.
+        @type signal_name: str
+        @param receiver: A handler to call when the signal is sent out.
+        """
+        self.signals[signal_name].append(receiver)
+
+    def disconnect(self, signal_name, receiver):
+        """
+        Disconnect a receiver from a signal.
+
+        @param signal_name: The name of the signal to disconnect the receiver from.
+        @type signal_name: str
+        @param receiver: The receiver to disconnect from the signal.
+        """
+        try:
+            self.signals[signal_name].remove(receiver)
+        except ValueError:
+            pass
+

tests/behavioral_tests/test_mediator.py

@@ -0,0 +1,97 @@
+from unittest import TestCase
+from pypatterns.behavioral.mediator import Mediator
+
+
+class MediatorTestCase(TestCase):
+    """
+    Unit testing class for the Mediator class.
+    """
+    def setUp(self):
+        """
+        Initialize testing data.
+        """
+        class Dog(object):
+            self.sound = ''
+
+            def set_sound(self, sound):
+                self.sound = sound
+
+        class Cat(object):
+            self.sound = ''
+
+            def set_sound(self, sound):
+                self.sound = sound
+
+        self.dog = Dog()
+        self.cat = Cat()
+
+    def test_connect(self):
+        """
+        Test connecting a receiver to a signal.
+
+        @raise AssertionError: If the test fails.
+        """
+        mediator = Mediator()
+        mediator.connect('set_dog_sound', self.dog.set_sound)
+        self.assertEquals([self.dog.set_sound], mediator.signals['set_dog_sound'])
+
+        mediator.connect('set_cat_sound', self.cat.set_sound)
+        self.assertEquals([self.cat.set_sound], mediator.signals['set_cat_sound'])
+
+    def test_disconnect(self):
+        """
+        Test disconnecting a receiver from a signal.
+
+        @raise AssertionError: If the test fails.
+        """
+        mediator = Mediator()
+        mediator.connect('set_dog_sound', self.dog.set_sound)
+        self.assertEquals([self.dog.set_sound], mediator.signals['set_dog_sound'])
+
+        mediator.disconnect('set_dog_sound', self.dog.set_sound)
+        self.assertEquals([], mediator.signals['set_dog_sound'])
+
+    def test_signal(self):
+        """
+        Test the signal method.
+
+        @raise AssertionError: If the test fails.
+        """
+        mediator = Mediator()
+        mediator.connect('set_dog_sound', self.dog.set_sound)
+        mediator.connect('set_cat_sound', self.cat.set_sound)
+        mediator.signal('set_dog_sound', 'woof')
+        mediator.signal('set_cat_sound', 'meow')
+
+        self.assertEquals('woof', self.dog.sound)
+        self.assertEquals('meow', self.cat.sound)
+
+    def test_invalid_disconnect(self):
+        """
+        Test disconnecting an unconnected receiver.
+
+        @raise AssertionError: If the test fails.
+        """
+        mediator = Mediator()
+        try:
+            mediator.disconnect('foo', self.dog.set_sound)
+            mediator.disconnect('bar', self.cat.set_sound)
+        except:
+            raise AssertionError()
+
+    def test_invalid_signal(self):
+        """
+        Test sending a signal that no one is connected to.
+
+        @raise AssertionError: If the test fails.
+        """
+        mediator = Mediator()
+
+        try:
+            mediator.signal('foo')
+        except:
+            raise AssertionError()
+
+
+
+