instrumentation/next-share/BaseLib/Test/test_multicast.py

   1 # Written by Njaal Borch
   2 # see LICENSE.txt for license information
   3 #
   4
   5 # Arno, 2009-04-16:
   6 # - You should also test whether Tribler responds correctly to multicast
   7 #   messages sent directly from a socket, and not via your code.
   8 #
   9 # - Some IPv6 tests will fail on Win32 if IPv6 is not installed.
  10 #
  11 #
  12
  13 import unittest
  14 import tempfile
  15 import select
  16
  17 from BaseLib.Core.Multicast import *
  18
  19
  20 # class MyLoggerTest(unittest.TestCase):
  21
  22 #     """
  23 #     Test the MyLogger class
  24
  25 #     """
  26
  27 #     def setUp(self):
  28 #         self.log = MyLogger()
  29 #         self.conn = BTConnection('localhost',self.hisport)
  30
  31 #     def testLog(self):
  32
  33 #         self.log.debug("DEBUG message")
  34
  35 #         self.log.info("INFO message")
  36
  37 #         self.log.warning("WARNING message")
  38
  39 #         self.log.fatal("FATAL message")
  40
  41 #         try:
  42 #             raise Exception("Exception text")
  43 #         except:
  44 #             self.log.exception("Should have a traceback below here:")
  45
  46
  47 class FakeOverlayBridge:
  48     def add_task(self, function, data):
  49         function()
  50
  51 class FakePeerDBHandler:
  52
  53     def addPeer(self, *args, **kargs):
  54         pass
  55
  56     def setPeerLocalFlag(self, *args):
  57         pass
  58
  59
  60 class TestUDPServer(threading.Thread):
  61
  62     def __init__(self, socket, mc_channel):
  63         threading.Thread.__init__(self)
  64
  65         self.socket = socket
  66         self.mc_channel = mc_channel
  67         self.running = True
  68
  69     def run(self):
  70
  71         while self.running:
  72             try:
  73                 if select.select([self.socket],[],[], 1)[0]:
  74                     (data, addr) = self.socket.recvfrom(1500)
  75                     self.mc_channel.data_came_in(addr, data)
  76             except Exception,e:
  77                 print e
  78
  79
  80     def stop(self):
  81         self.running = False
  82
  83
  84 class MulticastTest(unittest.TestCase):
  85
  86     """
  87     Test multicast class
  88
  89     """
  90
  91     def __init__(self, param):
  92         unittest.TestCase.__init__(self, param)
  93
  94         #TestAsServer.__init__(self, param)
  95         self.test_server = None
  96         self.overlay_bridge = FakeOverlayBridge()
  97         self.peer_db = FakePeerDBHandler()
  98
  99     def prepare_test(self, config, capabilitites=None):
 100         """
 101         Cannot be done by setUp as we need special config
 102         """
 103
 104         self.multicast = Multicast(config, self.overlay_bridge, 1234, 1, self.peer_db,
 105                                    capabilities=capabilitites)
 106
 107         self.test_server = TestUDPServer(self.multicast.getSocket(),
 108                                          self.multicast)
 109         self.test_server.start()
 110
 111     def tearDown(self):
 112         if self.test_server is not None:
 113             self.test_server.stop()
 114         self.multicast = None
 115
 116
 117     def testIPv4(self):
 118
 119         # Dummy config
 120         config = {'permid':'123',
 121                   'hostname':'myhostname',
 122                   'port':'1234',
 123                   'multicast_ipv4_address':'224.0.1.43',
 124                   'multicast_ipv6_address':'ff02::4124:1261:ffef',
 125                   'multicast_port':'6124',
 126                   'multicast_enabled':True,
 127                   'multicast_ipv4_enabled':True,
 128                   'multicast_ipv6_enabled':False,
 129                   'multicast_announce':True}
 130
 131         self.prepare_test(config)
 132
 133         failed = True
 134         seen = 0
 135         for (permid, addr, capabilities) in self.multicast.discoverNodes():
 136             if permid == '123':
 137                 failed = False
 138         if failed:
 139             raise Exception("Didn't discover myself using IPv4")
 140
 141
 142     def testIPv6(self):
 143
 144         # Dummy config
 145         config = {'permid':'123',
 146                   'multicast_ipv4_address':'224.0.1.43',
 147                   'multicast_ipv6_address':'ff02::4124:1261:ffef',
 148                   'multicast_port':'6124',
 149                   'multicast_enabled':True,
 150                   'multicast_ipv4_enabled':False,
 151                   'multicast_ipv6_enabled':True,
 152                   'multicast_announce':True}
 153
 154         self.prepare_test(config)
 155         failed = True
 156         for (permid, addr, capabilities) in self.multicast.discoverNodes():
 157             if permid == '123':
 158                 failed = False
 159         if failed:
 160             raise Exception("Didn't discover myself using IPv6")
 161
 162     def testBoth(self):
 163
 164         # Dummy config
 165         config = {'permid':'123',
 166                   'multicast_ipv4_address':'224.0.1.43',
 167                   'multicast_ipv6_address':'ff02::4124:1261:ffef',
 168                   'multicast_port':'6124',
 169                   'multicast_enabled':True,
 170                   'multicast_ipv4_enabled':True,
 171                   'multicast_ipv6_enabled':True,
 172                   'multicast_announce':True}
 173
 174         self.prepare_test(config)
 175
 176         seen = 0
 177         for (permid, addr, capabilities) in self.multicast.discoverNodes():
 178             if permid == '123':
 179                 seen += 1
 180         if seen < 2:
 181             raise Exception("Didn't discover myself enough using both (saw me %d times, expected 2)"%seen)
 182
 183     def testAllDisabled(self):
 184
 185         # Dummy config
 186         config = {'permid':'123',
 187                   'multicast_ipv4_address':'224.0.1.43',
 188                   'multicast_ipv6_address':'ff02::4124:1261:ffef',
 189                   'multicast_port':'6124',
 190                   'multicast_ipv4_enabled':False,
 191                   'multicast_ipv6_enabled':False,
 192                   'multicast_announce':True}
 193
 194         self.prepare_test(config)
 195
 196         try:
 197             if len(self.multicast.discoverNodes()) > 0:
 198                 raise Exception("Discovered nodes even though multicast is not allowed")
 199         except:
 200             # Expected
 201             pass
 202
 203
 204     def testAnnounce(self):
 205
 206         # Dummy config
 207         config = {'permid':'123',
 208                   'multicast_ipv4_address':'224.0.1.43',
 209                   'multicast_ipv6_address':'ff02::4124:1261:ffef',
 210                   'multicast_port':'6124',
 211                   'multicast_enabled':True,
 212                   'multicast_ipv4_enabled':True,
 213                   'multicast_ipv6_enabled':True,
 214                   'multicast_announce':True}
 215
 216         self.prepare_test(config)
 217
 218         # Handle the announce
 219         self.announces = []
 220         self.multicast.addAnnounceHandler(self.handleAnnounce)
 221         self.multicast.sendAnnounce(['elem1','elem2'])
 222
 223         # Wait for asynchronous handling
 224         time.sleep(2.0)
 225
 226         for announce in self.announces:
 227             if announce == ['123', 'elem1', 'elem2']:
 228                 return # Got it
 229
 230         raise Exception("Failed to get announce")
 231
 232     def handleAnnounce(self, permid, addr, list):
 233
 234         """
 235         Handle announce callback function
 236         """
 237         self.announces.append([permid] + list)
 238
 239     def testCapabilities(self):
 240         """
 241         Test capabilities thingy
 242         """
 243
 244         myCapabilities = ["Something", "something else", "something totally different"]
 245
 246         # Dummy config
 247         config = {'permid':'testCapabilities',
 248                   'multicast_ipv4_address':'224.0.1.43',
 249                   'multicast_ipv6_address':'ff02::4124:1261:ffef',
 250                   'multicast_port':'6124',
 251                   'multicast_enabled':True,
 252                   'multicast_ipv4_enabled':False,
 253                   'multicast_ipv6_enabled':True,
 254                   'multicast_announce':True}
 255
 256         self.prepare_test(config, myCapabilities)
 257
 258         failed = True
 259         for (permid, addr, capabilities) in self.multicast.discoverNodes():
 260             if permid == config['permid']:
 261                 failed = False
 262                 if capabilities != myCapabilities:
 263                     raise Exception("Got bad capabilities, got %s, expected %s"%(str(capabilities), str(myCapabilities)))
 264
 265         if failed:
 266             raise Exception("Didn't discover myself using IPv6")
 267
 268
 269 def test_suite():
 270     suite = unittest.TestSuite()
 271     suite.addTest(unittest.makeSuite(MulticastTest))
 272
 273     return suite
 274
 275
 276
 277 if __name__ == "__main__":
 278
 279     # TODO: Multicast does gives us multiple hits for ourselves, is that ok?
 280
 281     print "Testing the Multicast classes"
 282
 283     unittest.main()
 284
 285     print "All done"
 286