test_rmq.py 文件源码

def setUp(self):
        super(TestTaskControllerAndRunner, self).setUp()

        try:
            connection = pika.BlockingConnection()
        except pika.exceptions.ConnectionClosed:
            self.fail("Couldn't open connection.  Make sure rmq server is running")

        queue = "{}.{}.tasks".format(self.__class__, uuid.uuid4())
        self.sender = TaskController(connection, queue=queue)
        self.runner = TaskRunner(connection, queue=queue)