test/unit/pubsub.py

   1 #   test.pubsub - tests for publish/subscribe classes
   2 #
   3 #   Part of WiFi Radar: A utility for managing WiFi profiles on GNU/Linux.
   4 #
   5 #   Copyright (C) 2014 Sean Robinson <robinson@tuxfamily.org>
   6 #
   7 #   This program is free software; you can redistribute it and/or modify
   8 #   it under the terms of the GNU General Public License as published by
   9 #   the Free Software Foundation; version 2 of the License.
  10 #
  11 #   This program is distributed in the hope that it will be useful,
  12 #   but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 #   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 #   GNU General Public License in LICENSE.GPL for more details.
  15 #
  16 #   You should have received a copy of the GNU General Public License
  17 #   along with this program; if not, write to:
  18 #      Free Software Foundation, Inc.
  19 #      51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA
  20 #
  21
  22
  23 from __future__ import unicode_literals
  24
  25 from multiprocessing import Pipe
  26 import unittest
  27
  28 import mock
  29
  30 import wifiradar.pubsub as pubsub
  31
  32
  33 class TestLimitedDispatcher(unittest.TestCase):
  34     """ The Dispatcher tested here does not process messages. """
  35
  36     def setUp(self):
  37         self.dispatch = pubsub.Dispatcher(auto_start=False)
  38
  39     def test_subscribe(self):
  40         """ Test subscribe method. """
  41         sub01 = self.dispatch.subscribe()
  42         # There should be only one subscriber in the Dispatcher.
  43         self.assertEqual(1, len(self.dispatch.pipes))
  44
  45     def test_unsubscribe(self):
  46         """ Test unsubscribe method. """
  47         # Simulate the two ends of a multiprocessing.Pipe.
  48         a, b = mock.Mock(), mock.Mock()
  49
  50         self.dispatch.pipes[a] = ['TOPIC', 'EXIT']
  51         self.dispatch._pairs[b] = a
  52         # There should be only one subscriber in the Dispatcher.
  53         self.assertEqual(1, len(self.dispatch.pipes))
  54
  55         self.dispatch.unsubscribe(b)
  56         # There should be no subscribers in the Dispatcher.
  57         self.assertEqual(0, len(self.dispatch.pipes))
  58         # Check that both ends of the Pipe have been closed.
  59         a.close.assert_called_once_with()
  60         b.close.assert_called_once_with()
  61
  62     def tearDown(self):
  63         self.dispatch.close()
  64
  65
  66 class TestDispatcher(unittest.TestCase):
  67     def setUp(self):
  68         self.dispatch = pubsub.Dispatcher()
  69
  70     def test_simple_msg(self):
  71         """ Test sending and receiving a message. """
  72         sub01 = self.dispatch.subscribe('TEST')
  73         sub02 = self.dispatch.subscribe('TEST')
  74         msg = pubsub.Message('TEST', 'Hello')
  75         sub01.send(msg)
  76         self.assertEqual(True, sub02.poll(0.25))
  77         recv_msg = sub02.recv()
  78         self.assertEqual(msg.topic, recv_msg.topic)
  79         self.assertEqual(msg.details, recv_msg.details)
  80         # Subscribers do not get a copy of their own messages.
  81         self.assertEqual(False, sub01.poll(0.25))
  82
  83     def test_multi_msg(self):
  84         """ Test sending and receiving multiple messages. """
  85         subs = list()
  86         subs.append(self.dispatch.subscribe('TEST01'))
  87         subs.append(self.dispatch.subscribe('TEST02'))
  88         subs.append(self.dispatch.subscribe(('TEST01', 'TEST02')))
  89         subs.append(self.dispatch.subscribe())
  90
  91         msgs = list()
  92         msgs.append(pubsub.Message('TEST01', 'Hello'))
  93         msgs.append(pubsub.Message('TEST02', 'Hello'))
  94         msgs.append(pubsub.Message('TEST03', 'Hello'))
  95
  96         publisher = self.dispatch.subscribe()
  97         for m in msgs:
  98             publisher.send(m)
  99
 100         self.assertEqual(True, subs[0].poll(0.25))
 101         recv_msg = subs[0].recv()
 102         self.assertEqual(msgs[0].topic, recv_msg.topic)
 103         self.assertEqual(msgs[0].details, recv_msg.details)
 104         # No more messages waiting.
 105         self.assertEqual(False, subs[0].poll(0.25))
 106         self.assertEqual(True, subs[1].poll(0.25))
 107         recv_msg = subs[1].recv()
 108         self.assertEqual(msgs[1].topic, recv_msg.topic)
 109         self.assertEqual(msgs[1].details, recv_msg.details)
 110         # No more messages waiting.
 111         self.assertEqual(False, subs[1].poll(0.25))
 112         # Check for two messages, one for each subscribed topic.
 113         self.assertEqual(True, subs[2].poll(0.25))
 114         recv_msg = subs[2].recv()
 115         self.assertEqual(msgs[0].topic, recv_msg.topic)
 116         self.assertEqual(msgs[0].details, recv_msg.details)
 117         self.assertEqual(True, subs[2].poll(0.25))
 118         recv_msg = subs[2].recv()
 119         self.assertEqual(msgs[1].topic, recv_msg.topic)
 120         self.assertEqual(msgs[1].details, recv_msg.details)
 121         # No more messages waiting.
 122         self.assertEqual(False, subs[2].poll(0.25))
 123         # This subscriber should not have received any messages.
 124         self.assertEqual(False, subs[3].poll(0.25))
 125
 126         # Subscribers do not get a copy of their own messages.
 127         self.assertEqual(False, publisher.poll(0.25))
 128
 129     def tearDown(self):
 130         self.dispatch.close()
 131
 132
 133 class TestConnectors(unittest.TestCase):
 134     def setUp(self):
 135         self.local_dispatch = pubsub.Dispatcher()
 136         self.foreign_dispatch = pubsub.Dispatcher()
 137
 138     def test_connector(self):
 139         """ Test sending through a chain of Dispatchers. """
 140         msg = pubsub.Message('TEST', 'Hello')
 141         # There should be no subscribers in the Dispatchers.
 142         self.assertEqual(0, len(self.local_dispatch.pipes))
 143         self.assertEqual(0, len(self.foreign_dispatch.pipes))
 144         # Setup a Pipe connecting the two Dispatchers.
 145         a, b = Pipe()
 146         self.local_dispatch.add_connector(a)
 147         self.foreign_dispatch.add_connector(b)
 148         # There should be one subscriber in the Dispatchers.
 149         self.assertEqual(1, len(self.local_dispatch.pipes))
 150         self.assertEqual(1, len(self.foreign_dispatch.pipes))
 151         # Subscribe to 'TEST' in both Dispatchers.
 152         local_sub = self.local_dispatch.subscribe('TEST')
 153         foreign_sub = self.foreign_dispatch.subscribe('TEST')
 154         # There should be two subscribers in the Dispatchers.
 155         self.assertEqual(2, len(self.local_dispatch.pipes))
 156         self.assertEqual(2, len(self.foreign_dispatch.pipes))
 157         # Send test message...
 158         local_sub.send(msg)
 159         # ...receive test message.
 160         self.assertEqual(True, foreign_sub.poll(0.25))
 161         recv_msg = foreign_sub.recv()
 162         self.assertEqual(msg.topic, recv_msg.topic)
 163         self.assertEqual(msg.details, recv_msg.details)
 164         # Remove link between the two Dispatchers.
 165         self.local_dispatch.remove_connector(a)
 166         self.foreign_dispatch.remove_connector(b)
 167         # Unsubscribe from both Dispatchers.
 168         self.local_dispatch.unsubscribe(local_sub)
 169         self.foreign_dispatch.unsubscribe(foreign_sub)
 170         # There should be no subscribers in the Dispatchers.
 171         self.assertEqual(0, len(self.local_dispatch.pipes))
 172         self.assertEqual(0, len(self.foreign_dispatch.pipes))
 173
 174     def test_bridge(self):
 175         """ Test the bridge function. """
 176         # There should be no subscribers in the Dispatchers.
 177         self.assertEqual(0, len(self.local_dispatch.pipes))
 178         self.assertEqual(0, len(self.foreign_dispatch.pipes))
 179         # Create the link...
 180         local, foreign = pubsub.bridge(self.local_dispatch,
 181                                        self.foreign_dispatch)
 182         # There should be one subscriber in the Dispatchers.
 183         self.assertEqual(1, len(self.local_dispatch.pipes))
 184         self.assertEqual(1, len(self.foreign_dispatch.pipes))
 185         # Remove the link.
 186         self.local_dispatch.remove_connector(local)
 187         self.foreign_dispatch.remove_connector(foreign)
 188         # There should be no subscribers in the Dispatchers.
 189         self.assertEqual(0, len(self.local_dispatch.pipes))
 190         self.assertEqual(0, len(self.foreign_dispatch.pipes))
 191
 192     def tearDown(self):
 193         self.local_dispatch.close()
 194         self.foreign_dispatch.close()