test_messaging.py 文件源码

def test_double_subscriptions(self):
        """
        Ensure that messages are delivered to all subscriptions of a topic.
        (e.g. identifies queue setup problems)
        :return:
        """
        self.m.subscribe(self._simple_subscribe_cbf1, "test.interleave")
        self.m.subscribe(self._simple_subscribe_cbf2, "test.interleave")
        time.sleep(0.5)
        # send publish to notify endpoint
        self.m.publish("test.interleave", "my-notification1")
        # enusre that it is received by each subscription
        self.assertTrue(self.wait_for_particular_messages("my-notification1", buffer=0))
        self.assertTrue(self.wait_for_particular_messages("my-notification1", buffer=1))

    #@unittest.skip("disabled")