test_status.py 文件源码

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

        self.response = None

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

        exchange = "{}.{}.status_request".format(self.__class__, uuid.uuid4())
        self.requester = StatusRequester(self._connection, exchange=exchange)
        self.manager = ProcessManager()
        self.provider = StatusProvider(
            self._connection, process_manager=self.manager, exchange=exchange)