test.py 文件源码

def test_multihop(self):
        random.shuffle(self.swarm)
        senders = self.swarm[:len(self.swarm) / 2]
        receivers = self.swarm[len(self.swarm) / 2:]
        for sender, receiver in zip(senders, receivers):

            # receiver subscribes to topic
            topic = "test_miltihop_{0}".format(binascii.hexlify(os.urandom(32)))
            receiver.pubsub_subscribe(topic)

            # wait until subscriptions propagate
            time.sleep(SLEEP_TIME)

            # send event
            event = binascii.hexlify(os.urandom(32))
            sender.pubsub_publish(topic, event)

            # wait until event propagates
            time.sleep(SLEEP_TIME)

            # check all peers received the event
            events = receiver.pubsub_events(topic)
            self.assertEqual(events, [event])