test_integration.py 文件源码

def test_topic_runs_callback(self):
        await self.GIVEN_ProducerRegistered(exchange_name="fake_topic_exch",
                                      exchange_type="topic")

        await self.GIVEN_ConsumerRegistered(queue_name="fake_topic_consumer_queue",
                                      exchange_name="fake_topic_exch",
                                      exchange_type="topic",
                                      routing_keys=["*.red","ball.*"],
                                      callback = self.fake_callback)

        await self.GIVEN_MessagePublished(exchange_name="fake_topic_exch",
                                    msg="fake_message",
                                    routing_key="apple.red")

        await self.WHEN_ProcessEventsOnce()

        self.THEN_CallbackReceivesMessage("fake_message")

    # @unittest.skip("skipped")